polyester-sdk 0.1.0-alpha.28

//! POLY-3746 L2 integration tests via public SDK APIs + local mock HTTP/WS.
//!
//! These exercise production paths (subscribe_raw, JsonRpcClient, wait_for_catalogs,
//! Quantity::format, SnapshotThenStream) against in-process servers — not private helpers.

mod hardening_support;

use buffa::Message;
use futures_util::future::FutureExt;
use hardening_support::{
    GET_BALANCES_PATH, GET_ORDER_PATH, GET_TRADES_PATH, MockHttpServer, MockWsServer,
    SPOT_CONFIG_PATH, ZIPPER_CONFIG_PATH, centrifugo_ok_reply, connect_proto_ok, test_credentials,
    wait_until,
};
use polyester::Error;
use polyester::auth::Credentials;
use polyester::chain::JsonRpcClient;
use polyester::codecs::{MAX_PROTOCOL_SCALE, format_ledger_u128, format_qty_scaled};
use polyester::realtime::Client as RealtimeClient;
use polyester::realtime::{
    MAX_REALTIME_MESSAGE_BYTES, SnapshotThenStream, SnapshotThenStreamConfig,
};
use polyester::transport::MAX_CONNECT_RESPONSE_BYTES;
use polyester::{Client, Config, Quantity, QuantityDomain};
use serde_json::json;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

const TEST_KEY: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
const PRIVATE_CHANNEL: &str = "private:spot:orders:acct:proto";
const PUBLIC_CHANNEL: &str = "public:spot:market:trades:1:proto";
type PublicationDecoder =
    Arc<dyn Fn(&[u8]) -> polyester::Result<()> + Send + Sync + std::panic::RefUnwindSafe>;

fn publication_decoders() -> Vec<(&'static str, PublicationDecoder)> {
    use polyester::codecs::decode::{
        account_identity_from_bytes, address_book_invalidation_from_bytes, api_key_from_bytes,
        api_policy_from_bytes, asset_balance_from_bytes, candle_point_from_bytes,
        flow_detail_from_bytes, flow_summary_from_bytes, heatmap_live_bucket_from_bytes,
        ledger_transfer_from_bytes, market_overview_batch_from_bytes, market_trade_from_bytes,
        order_from_bytes, orderbook_delta_from_bytes, subaccount_from_bytes,
        subaccount_policy_from_bytes, trigger_event_from_bytes, trigger_from_bytes,
        user_trade_from_bytes, zipped_asset_supply_batch_from_bytes,
    };

    macro_rules! decoder {
        ($name:literal, $decode:expr) => {
            (
                $name,
                Arc::new(move |bytes: &[u8]| ($decode)(bytes).map(|_| ())) as PublicationDecoder,
            )
        };
    }

    vec![
        decoder!("order", order_from_bytes),
        decoder!("user_trade", user_trade_from_bytes),
        decoder!("asset_balance", asset_balance_from_bytes),
        decoder!("ledger_transfer", ledger_transfer_from_bytes),
        decoder!("trigger", trigger_from_bytes),
        decoder!("trigger_event", trigger_event_from_bytes),
        decoder!("market_trade", market_trade_from_bytes(8)),
        decoder!("orderbook_delta", orderbook_delta_from_bytes),
        decoder!("flow_summary", flow_summary_from_bytes),
        decoder!("flow_detail", flow_detail_from_bytes),
        decoder!("account_identity", account_identity_from_bytes),
        decoder!(
            "candle_point",
            candle_point_from_bytes(1, "1m".to_owned(), 8)
        ),
        decoder!("heatmap_live_bucket", heatmap_live_bucket_from_bytes),
        decoder!("market_overview_batch", market_overview_batch_from_bytes),
        decoder!("zipped_asset_supply_batch", |bytes| {
            zipped_asset_supply_batch_from_bytes(bytes, |_| Some(18))
        }),
        decoder!("api_key", api_key_from_bytes),
        decoder!("subaccount", subaccount_from_bytes),
        decoder!("subaccount_policy", subaccount_policy_from_bytes),
        decoder!("api_policy", api_policy_from_bytes),
        decoder!(
            "address_book_invalidation",
            address_book_invalidation_from_bytes
        ),
    ]
}

#[test]
fn l2_all_publication_decoders_are_panic_free_for_adversarial_protobuf() {
    let decoders = publication_decoders();
    let fixed_corpus: &[&[u8]] = &[
        &[],
        &[0x0f],
        &[0x0a],
        &[0x0a, 0xff],
        &[0x80; 11],
        &[0xff; 32],
        &[0x0a, 0xff, 0xff, 0xff, 0xff, 0x0f],
    ];

    for (name, decode) in &decoders {
        for payload in fixed_corpus {
            let result = std::panic::catch_unwind(|| decode(payload));
            assert!(result.is_ok(), "{name} panicked for {payload:02x?}");
            assert!(
                result.unwrap().is_err(),
                "{name} accepted malformed protobuf {payload:02x?}"
            );
        }
    }

    // Deterministic corpus: 5,000 arbitrary byte strings and 5,000
    // structurally well-framed protobuf messages, each exercised against all
    // 20 publication decoders (200,000 decoder invocations). The latter gets
    // beyond the first tag and into nested/manual conversion logic.
    let mut state = 0x9e37_79b9_7f4a_7c15_u64;
    for case in 0..5_000_usize {
        state ^= state << 13;
        state ^= state >> 7;
        state ^= state << 17;
        let len = 1 + (usize::try_from(state).unwrap_or(usize::MAX) % 512);
        let mut payload = vec![0_u8; len];
        for byte in &mut payload {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            *byte = state as u8;
        }
        for (name, decode) in &decoders {
            let result = std::panic::catch_unwind(|| decode(&payload));
            assert!(result.is_ok(), "{name} panicked for mutation case {case}");
        }
    }
    for case in 0..5_000_usize {
        let mut payload = Vec::new();
        state ^= state << 13;
        state ^= state >> 7;
        state ^= state << 17;
        let fields = 1 + (state as usize % 8);
        for _ in 0..fields {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            let field_number = 1 + (state as u8 % 20);
            match state % 3 {
                0 => {
                    payload.push(field_number << 3);
                    let mut value = state;
                    loop {
                        let byte = (value & 0x7f) as u8;
                        value >>= 7;
                        if value == 0 {
                            payload.push(byte);
                            break;
                        }
                        payload.push(byte | 0x80);
                    }
                }
                1 => {
                    payload.push((field_number << 3) | 1);
                    payload.extend_from_slice(&state.to_le_bytes());
                }
                _ => {
                    payload.push((field_number << 3) | 2);
                    let len = state as usize % 24;
                    payload.push(len as u8);
                    for _ in 0..len {
                        state ^= state << 13;
                        state ^= state >> 7;
                        state ^= state << 17;
                        payload.push(state as u8);
                    }
                }
            }
        }
        for (name, decode) in &decoders {
            let result = std::panic::catch_unwind(|| decode(&payload));
            assert!(
                result.is_ok(),
                "{name} panicked for well-framed mutation case {case}"
            );
        }
    }

    let oversized = vec![0_u8; MAX_REALTIME_MESSAGE_BYTES + 1];
    for (name, decode) in &decoders {
        let error = decode(&oversized).expect_err("oversized publication must fail closed");
        assert!(
            error.to_string().contains("exceeds"),
            "{name} returned the wrong oversized-payload error: {error}"
        );
    }
}

#[tokio::test]
async fn l2_all_publication_decode_errors_surface_through_the_public_websocket_api() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_publication_after_handshake(active.clone(), vec![0x0f])
        .await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);

    for (name, decode) in publication_decoders() {
        let mut sub = rt
            .subscribe_proto_with_options(PUBLIC_CHANNEL, move |bytes| decode(bytes), false)
            .await
            .unwrap_or_else(|err| panic!("{name} handshake failed: {err}"));
        let error = tokio::time::timeout(Duration::from_secs(2), sub.recv_result())
            .await
            .unwrap_or_else(|_| panic!("{name} decode failure did not terminate the feed"))
            .unwrap_err();
        assert!(
            error.to_string().contains("proto decode"),
            "{name} returned the wrong error: {error}"
        );
        assert!(
            !sub.is_alive(),
            "{name} feed remained alive after decode error"
        );
        sub.close();
    }

    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_secs(2),
    )
    .await;
}

#[tokio::test(flavor = "current_thread")]
async fn l2_auth_10k_identical_requests_are_unique_future_bounded_and_runtime_safe() {
    use polyester::auth::{HEADER_SIGNATURE, HEADER_TIMESTAMP, MAX_SIGNING_FUTURE_SKEW_MS};
    use std::collections::HashSet;
    use std::time::{SystemTime, UNIX_EPOCH};

    let credentials = test_credentials("ak_test", TEST_KEY);
    let ticker = tokio::spawn(async {
        let mut previous = tokio::time::Instant::now();
        let mut largest_gap = Duration::ZERO;
        for _ in 0..100 {
            tokio::time::sleep(Duration::from_millis(10)).await;
            let now = tokio::time::Instant::now();
            largest_gap = largest_gap.max(now.duration_since(previous));
            previous = now;
        }
        largest_gap
    });
    // Let the ticker arm its first timer before creating 10k runnable tasks so
    // task-spawn overhead is not misclassified as signing backpressure.
    tokio::task::yield_now().await;
    let handles = (0..10_000)
        .map(|_| {
            let credentials = credentials.clone();
            tokio::spawn(async move {
                let headers = credentials
                    .sign_request_async(
                        "POST",
                        "https://api.example.test/orders.v1.OrdersService/CreateOrder",
                        b"identical-order-body",
                        None,
                    )
                    .await
                    .unwrap();
                (
                    headers[HEADER_TIMESTAMP].parse::<u64>().unwrap(),
                    headers[HEADER_SIGNATURE].clone(),
                    u64::try_from(
                        SystemTime::now()
                            .duration_since(UNIX_EPOCH)
                            .unwrap()
                            .as_millis(),
                    )
                    .unwrap(),
                )
            })
        })
        .collect::<Vec<_>>();
    let mut tuples = Vec::with_capacity(handles.len());
    for handle in handles {
        tuples.push(handle.await.unwrap());
    }
    let largest_timer_gap = ticker.await.unwrap();
    assert_eq!(tuples.len(), 10_000);
    assert!(
        largest_timer_gap < Duration::from_secs(2),
        "signing backpressure stalled current-thread Tokio for {largest_timer_gap:?}"
    );
    let timestamps = tuples.iter().map(|item| item.0).collect::<HashSet<_>>();
    let signatures = tuples.iter().map(|item| &item.1).collect::<HashSet<_>>();
    assert_eq!(timestamps.len(), 10_000, "duplicate signing timestamp");
    assert_eq!(signatures.len(), 10_000, "duplicate authentication tuple");
    for (timestamp, _, observed_at_ms) in &tuples {
        assert!(
            *timestamp <= *observed_at_ms + MAX_SIGNING_FUTURE_SKEW_MS,
            "signed timestamp {timestamp} exceeded its per-request bounded ceiling {}",
            *observed_at_ms + MAX_SIGNING_FUTURE_SKEW_MS
        );
    }

    let later = credentials
        .sign_request_async(
            "GET",
            "https://api.example.test/auth.v1.AuthService/Me",
            b"",
            None,
        )
        .await
        .unwrap()[HEADER_TIMESTAMP]
        .parse::<u64>()
        .unwrap();
    let wall_clock_later = u64::try_from(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_millis(),
    )
    .unwrap();
    assert!(later <= wall_clock_later + MAX_SIGNING_FUTURE_SKEW_MS);
}

#[tokio::test]
async fn l2_scale_dependent_public_paths_fail_closed_without_catalogs() {
    let client = Client::new(Config {
        api_url: "http://127.0.0.1:1".into(),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();

    let orderbook_err = client
        .orderbook
        .get("ETH-USDT", Some(50))
        .await
        .expect_err("orderbook must not guess scale 8");
    assert!(matches!(orderbook_err, Error::Validation(_)));
    assert!(orderbook_err.to_string().contains("catalog quantity scale"));

    let trades_err = client
        .market_data
        .get_trades_with(polyester::models::GetTradesOpts {
            symbol_id: Some(2),
            ..Default::default()
        })
        .await
        .expect_err("explicit symbol id must still require a known scale");
    assert!(matches!(trades_err, Error::Validation(_)));
    assert!(trades_err.to_string().contains("catalog quantity scale"));
}

#[tokio::test]
async fn digit_only_public_subaccount_id_uses_canonical_base58_on_order_wire() {
    use buffa::Message as _;
    use polyester::models::{CreateOrderParams, CreateOrderType, CreateSide, CreateTimeInForce};
    use polyester::proto::orders::v1::{CreateOrderRequest, CreateOrderResponse};
    use polyester::{Price, Quantity};
    use std::sync::Mutex;

    let observed = Arc::new(Mutex::new(None));
    let observed_handler = observed.clone();
    let server = MockHttpServer::spawn(move |request| {
        if request.path == "/orders.v1.OrdersService/CreateOrder" {
            let decoded = CreateOrderRequest::decode_from_slice(&request.body).unwrap();
            *observed_handler.lock().unwrap() = decoded.subaccount_id;
            return connect_proto_ok(&CreateOrderResponse {
                order_id: 9,
                client_order_id: "scope-wire".into(),
                ..Default::default()
            });
        }
        hardening_support::HttpScript::NotFound
    })
    .await;
    let client = Client::new(Config {
        api_url: server.base_url(),
        api_key_id: Some("scope-test".into()),
        api_private_key: Some(TEST_KEY.into()),
        default_sub_account_id: Some("5".into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    client
        .catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "BTC-USDT",
                "symbol_id": 1,
                "base_quantity_scale": 8
            }]
        }))
        .unwrap();
    client
        .orders
        .create(CreateOrderParams {
            symbol: "BTC-USDT".into(),
            side: CreateSide::Buy,
            order_type: CreateOrderType::Limit,
            quantity: Some(
                Quantity::from_scaled(
                    1,
                    Some(8),
                    QuantityDomain::OrderBase,
                    Some("BTC-USDT".into()),
                    Some(1),
                )
                .unwrap(),
            ),
            max_quote_debit_scaled: None,
            price: Some(Price::from_ticks(1, Some("BTC-USDT".into())).unwrap()),
            time_in_force: Some(CreateTimeInForce::Gtc),
            client_order_id: Some("scope-wire".into()),
            subaccount_id: None,
            post_only: None,
            market_client_ref_price: None,
            fee_asset: None,
            self_trade_prevention: None,
            market_max_slippage: None,
            attached_risk: None,
        })
        .await
        .unwrap();

    assert_eq!(*observed.lock().unwrap(), Some(4));
}

#[tokio::test]
async fn singular_mutation_default_connect_responses_fail_closed() {
    use polyester::models::{
        CreateInternalTransferParams, CreateOrderParams, CreateOrderType, CreateSide,
        CreateTimeInForce, CreateTradingWithdrawParams, CreateTriggerParams, CreateTriggerType,
    };
    use polyester::proto::chain::withdraw::v1::CreateTradingWithdrawResponse;
    use polyester::proto::marketdata::v1::GetCandlesResponse;
    use polyester::proto::orders::v1::CreateOrderResponse;
    use polyester::proto::transfer::v1::CreateInternalTransferResponse;
    use polyester::proto::triggers::v1::CreateTriggerResponse;
    use polyester::{AssetAmount, Price, Quantity};

    let server = MockHttpServer::spawn(|request| match request.path.as_str() {
        "/orders.v1.OrdersService/CreateOrder" => connect_proto_ok(&CreateOrderResponse::default()),
        "/triggers.v1.TriggersService/CreateTrigger" => {
            connect_proto_ok(&CreateTriggerResponse::default())
        }
        "/chain.withdraw.v1.WithdrawService/CreateTradingWithdraw" => {
            connect_proto_ok(&CreateTradingWithdrawResponse::default())
        }
        "/transfer.v1.InternalTransferService/CreateInternalTransfer" => {
            connect_proto_ok(&CreateInternalTransferResponse::default())
        }
        "/marketdata.v1.MarketDataService/GetCandles" => {
            connect_proto_ok(&GetCandlesResponse::default())
        }
        _ => hardening_support::HttpScript::NotFound,
    })
    .await;
    let client = Client::new(Config {
        api_url: server.base_url(),
        api_key_id: Some("fault-test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    client
        .catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "BTC-USDT",
                "symbol_id": 1,
                "base_quantity_scale": 8
            }]
        }))
        .unwrap();

    let quantity = Quantity::from_scaled(
        1,
        Some(8),
        QuantityDomain::OrderBase,
        Some("BTC-USDT".into()),
        Some(1),
    )
    .unwrap();
    let price = Price::from_ticks(1, Some("BTC-USDT".into())).unwrap();
    let order_error = client
        .orders
        .create(CreateOrderParams {
            symbol: "BTC-USDT".into(),
            side: CreateSide::Buy,
            order_type: CreateOrderType::Limit,
            quantity: Some(quantity.clone()),
            max_quote_debit_scaled: None,
            price: Some(price.clone()),
            time_in_force: Some(CreateTimeInForce::Gtc),
            client_order_id: Some("fault-order".into()),
            subaccount_id: Some(4),
            post_only: None,
            market_client_ref_price: None,
            fee_asset: None,
            self_trade_prevention: None,
            market_max_slippage: None,
            attached_risk: None,
        })
        .await
        .unwrap_err();
    assert!(matches!(order_error, Error::ResponseContract { .. }));

    let trigger_error = client
        .triggers
        .create(CreateTriggerParams {
            symbol: "BTC-USDT".into(),
            trigger_type: CreateTriggerType::StopLoss,
            side: CreateSide::Sell,
            order_type: CreateOrderType::Market,
            qty: quantity,
            trigger_price: Some(price),
            limit_price: None,
            trigger_price_source: None,
            time_in_force: None,
            subaccount_id: Some(4),
            client_trigger_id: "fault-trigger".into(),
            post_only: false,
            activation_price: None,
            trailing_distance_ticks: None,
            trailing_distance_bps: None,
            max_slippage_ticks: None,
            max_slippage_bps: None,
            twap_duration_ms: None,
            twap_slice_interval_ms: None,
            ladder_price_min: None,
            ladder_price_max: None,
            ladder_levels: None,
            ladder_distribution: None,
            fee_asset: None,
            self_trade_prevention_mode: None,
        })
        .await
        .unwrap_err();
    assert!(matches!(trigger_error, Error::Transport(_)));

    let amount = AssetAmount::from_scaled(1, Some(18), QuantityDomain::LedgerE18, Some(7)).unwrap();
    let withdraw_error = client
        .withdraw
        .create_to_funding(CreateTradingWithdrawParams {
            asset_id: 7,
            amount: amount.clone(),
            payload_signature: vec![1],
            destination_address: String::new(),
            idempotency_key: "fault-withdraw".into(),
            amount_scale: Some(18),
            deadline_ts_sec: Some(1_800_000_000),
            nonce: 1,
        })
        .await
        .unwrap_err();
    assert!(matches!(withdraw_error, Error::Transport(_)));

    let transfer_error = client
        .internal_transfers
        .create(CreateInternalTransferParams {
            asset_id: 7,
            quantity: amount,
            idempotency_key: "fault-transfer".into(),
            subaccount_id: Some(4),
            destination_account_id: Some("2".into()),
            destination_subaccount_id: None,
            destination_smart_account_address: None,
            quantity_scale: Some(18),
        })
        .await
        .unwrap_err();
    assert!(matches!(transfer_error, Error::Transport(_)));

    assert!(
        client
            .market_data
            .get_current_candle("BTC-USDT", "1m")
            .await
            .unwrap()
            .is_none()
    );
}

#[tokio::test]
async fn l2_invalid_client_order_and_trigger_ids_fail_closed_without_hitting_connect() {
    use polyester::models::{
        CreateOrderParams, CreateOrderType, CreateSide, CreateTimeInForce, CreateTriggerParams,
        CreateTriggerType,
    };
    use polyester::{Price, Quantity, QuantityDomain};
    use serde_json::json;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let hits = Arc::new(AtomicUsize::new(0));
    let hits_cb = Arc::clone(&hits);
    let http = MockHttpServer::spawn(move |_req| {
        hits_cb.fetch_add(1, Ordering::SeqCst);
        hardening_support::HttpScript::NotFound
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .expect("client");
    client
        .catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "BTC-USDT",
                "symbol_id": 1,
                "base_quantity_scale": 8
            }]
        }))
        .unwrap();

    let quantity = Quantity::from_scaled(
        1,
        Some(8),
        QuantityDomain::OrderBase,
        Some("BTC-USDT".into()),
        Some(1),
    )
    .unwrap();
    let price = Price::from_ticks(1, Some("BTC-USDT".into())).unwrap();

    let order_err = client
        .orders
        .create(CreateOrderParams {
            symbol: "BTC-USDT".into(),
            side: CreateSide::Buy,
            order_type: CreateOrderType::Limit,
            quantity: Some(quantity.clone()),
            max_quote_debit_scaled: None,
            price: Some(price.clone()),
            time_in_force: Some(CreateTimeInForce::Gtc),
            client_order_id: Some("bad id!".into()),
            subaccount_id: None,
            post_only: None,
            market_client_ref_price: None,
            fee_asset: None,
            self_trade_prevention: None,
            market_max_slippage: None,
            attached_risk: None,
        })
        .await
        .expect_err("invalid client_order_id must fail locally");
    assert!(
        matches!(order_err, Error::Validation(_)),
        "want Validation, got {order_err:?}"
    );

    let trigger_err = client
        .triggers
        .create(CreateTriggerParams {
            symbol: "BTC-USDT".into(),
            trigger_type: CreateTriggerType::StopLoss,
            side: CreateSide::Sell,
            order_type: CreateOrderType::Market,
            qty: quantity,
            trigger_price: Some(price),
            limit_price: None,
            trigger_price_source: None,
            time_in_force: None,
            subaccount_id: None,
            client_trigger_id: "x".repeat(37),
            post_only: false,
            activation_price: None,
            trailing_distance_ticks: None,
            trailing_distance_bps: None,
            max_slippage_ticks: None,
            max_slippage_bps: None,
            twap_duration_ms: None,
            twap_slice_interval_ms: None,
            ladder_price_min: None,
            ladder_price_max: None,
            ladder_levels: None,
            ladder_distribution: None,
            fee_asset: None,
            self_trade_prevention_mode: None,
        })
        .await
        .expect_err("oversized client_trigger_id must fail locally");
    assert!(
        matches!(trigger_err, Error::Validation(_)),
        "want Validation, got {trigger_err:?}"
    );

    assert_eq!(
        hits.load(Ordering::SeqCst),
        0,
        "invalid correlation ids must not reach Connect"
    );
}

#[tokio::test]
async fn managed_overview_overflow_terminates_receiver_callback_task_and_socket() {
    use polyester::proto::marketoverview::v1::{
        ListMarketOverviewResponse, MarketOverview, MarketOverviewBatch,
    };
    use polyester::services::MarketOverviewCreateSubscriptionOptions;
    use std::sync::atomic::AtomicBool;

    let active = Arc::new(AtomicUsize::new(0));
    let publication = MarketOverviewBatch {
        markets: vec![MarketOverview {
            symbol_id: 1,
            symbol: "BTC-USDT".into(),
            last_price_ticks: 1,
            ..Default::default()
        }],
        ..Default::default()
    }
    .encode_to_vec();
    let ws = MockWsServer::spawn_centrifugo_publication_flood_after_handshake(
        active.clone(),
        publication,
        500,
    )
    .await;
    let http = MockHttpServer::spawn(|request| {
        if request.path == "/marketoverview.v1.MarketOverviewService/ListMarketOverview" {
            connect_proto_ok(&ListMarketOverviewResponse::default())
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        ws_url: ws.ws_url(),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    let mut subscription = client
        .market_overview
        .create_subscription(MarketOverviewCreateSubscriptionOptions::default())
        .await
        .unwrap();
    let callback_fired = Arc::new(AtomicBool::new(false));
    let callback_flag = callback_fired.clone();
    subscription.set_on_error(move |error| {
        assert!(matches!(error, Error::QueueOverflow(_)));
        callback_flag.store(true, Ordering::SeqCst);
    });

    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    while subscription.err().is_none() && tokio::time::Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert!(
        matches!(subscription.err(), Some(Error::QueueOverflow(_))),
        "unexpected terminal state: err={:?}, active={}, connects={}",
        subscription.err(),
        active.load(Ordering::SeqCst),
        ws.connects.load(Ordering::SeqCst)
    );
    loop {
        let next = tokio::time::timeout(Duration::from_secs(2), subscription.updates().recv())
            .await
            .expect("overflow must terminate the managed receiver");
        if next.is_none() {
            break;
        }
    }
    assert!(matches!(subscription.err(), Some(Error::QueueOverflow(_))));
    assert!(callback_fired.load(Ordering::SeqCst));
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_secs(2),
    )
    .await;

    // Explicit close remains idempotent after overflow and cannot leave the
    // already-terminated transport alive.
    subscription.close();
    assert_eq!(active.load(Ordering::SeqCst), 0);
}

#[tokio::test]
async fn managed_orderbook_overflow_terminates_receiver_callback_task_and_socket() {
    use polyester::proto::orderbook::v1::{GetOrderBookResponse, OrderBookDelta};
    use polyester::services::CreateSubscriptionOptions;
    use std::sync::atomic::AtomicBool;

    let active = Arc::new(AtomicUsize::new(0));
    let publication = OrderBookDelta {
        symbol_id: 1,
        book_seq_start: 2,
        book_seq_end: 2,
        ..Default::default()
    }
    .encode_to_vec();
    let ws = MockWsServer::spawn_centrifugo_publication_flood_after_handshake(
        active.clone(),
        publication,
        500,
    )
    .await;
    let http = MockHttpServer::spawn(|request| {
        if request.path == "/orderbook.v1.OrderbookService/GetOrderBook" {
            connect_proto_ok(&GetOrderBookResponse {
                symbol_id: 1,
                book_seq: 1,
                ..Default::default()
            })
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        ws_url: ws.ws_url(),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    client
        .catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "BTC-USDT",
                "symbol_id": 1,
                "base_quantity_scale": 8
            }]
        }))
        .unwrap();
    let mut subscription = client
        .orderbook
        .create_subscription(CreateSubscriptionOptions {
            symbol: "BTC-USDT".into(),
            symbol_id: Some(1),
            depth: Some(50),
            bucket: None,
        })
        .await
        .unwrap();
    let callback_fired = Arc::new(AtomicBool::new(false));
    let callback_flag = callback_fired.clone();
    subscription.set_on_error(move |error| {
        assert!(matches!(error, Error::QueueOverflow(_)));
        callback_flag.store(true, Ordering::SeqCst);
    });

    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    while subscription.err().is_none() && tokio::time::Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert!(
        matches!(subscription.err(), Some(Error::QueueOverflow(_))),
        "unexpected terminal state: err={:?}, active={}, connects={}",
        subscription.err(),
        active.load(Ordering::SeqCst),
        ws.connects.load(Ordering::SeqCst)
    );
    loop {
        let next = tokio::time::timeout(Duration::from_secs(2), subscription.updates().recv())
            .await
            .expect("overflow must terminate the managed receiver");
        if next.is_none() {
            break;
        }
    }
    assert!(callback_fired.load(Ordering::SeqCst));
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_secs(2),
    )
    .await;
    subscription.close();
    assert_eq!(active.load(Ordering::SeqCst), 0);
}

#[test]
fn l2_contradictory_catalog_refreshes_fail_atomically_through_public_manager() {
    let client = Client::new(Config {
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    client
        .catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "BTC-USDT",
                "symbol_id": 1,
                "base_quantity_scale": 8
            }]
        }))
        .unwrap();
    let err = client
        .catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [
                {"symbol": "ETH-USDT", "symbol_id": 2, "base_quantity_scale": 6},
                {"symbol": "SOL-USDT", "symbol_id": 2, "base_quantity_scale": 8}
            ]
        }))
        .unwrap_err();
    assert!(err.to_string().contains("symbol_id 2"));
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        Some(8)
    );
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("ETH-USDT"),
        None
    );
}

#[tokio::test]
async fn l2_rest_and_realtime_market_trades_attach_catalog_scale_metadata() {
    use polyester::proto::marketdata::v1::{GetTradesResponse, MarketTrade as ProtoMarketTrade};

    for (symbol, symbol_id, scale, expected) in
        [("ETH-USDT", 2_u32, 6_u32, "1"), ("BTC-USDT", 1, 8, "0.01")]
    {
        let proto_trade = ProtoMarketTrade {
            symbol_id,
            match_id: 7,
            qty_scaled: 1_000_000,
            price_ticks: 1_000_000,
            ..Default::default()
        };
        let response = GetTradesResponse {
            trades: vec![proto_trade.clone()],
            ..Default::default()
        };
        let http = MockHttpServer::spawn(move |req| {
            if req.path == GET_TRADES_PATH {
                connect_proto_ok(&response)
            } else {
                hardening_support::HttpScript::NotFound
            }
        })
        .await;
        let active = Arc::new(AtomicUsize::new(0));
        let ws = MockWsServer::spawn_centrifugo_publication_after_handshake(
            active,
            proto_trade.encode_to_vec(),
        )
        .await;
        let client = Client::new(Config {
            api_url: http.base_url(),
            ws_url: ws.ws_url(),
            hydrate_catalogs: false,
            timeout: Duration::from_secs(2),
            ..Default::default()
        })
        .unwrap();
        client
            .catalogs
            .hydrate_spot_config_json(json!({
                "pairs": [{
                    "symbol": symbol,
                    "symbol_id": symbol_id,
                    "base_quantity_scale": scale
                }]
            }))
            .unwrap();

        let rest = client
            .market_data
            .get_trades(symbol, Some(1))
            .await
            .unwrap();
        assert_eq!(
            rest.trades[0].qty.as_ref().unwrap().format(None).unwrap(),
            expected
        );

        let mut realtime = client.market_data.subscribe_trades(symbol).await.unwrap();
        let streamed = tokio::time::timeout(Duration::from_secs(2), realtime.recv_result())
            .await
            .expect("realtime publication deadline")
            .unwrap()
            .expect("realtime publication");
        assert_eq!(
            streamed.qty.as_ref().unwrap().format(None).unwrap(),
            expected
        );
        realtime.close();
    }
}

fn private_rt(ws: &MockWsServer, http: &MockHttpServer, timeout: Duration) -> RealtimeClient {
    RealtimeClient::with_timeout(
        ws.ws_url(),
        http.base_url(),
        Some(test_credentials("ak_test", TEST_KEY)),
        None,
        timeout,
    )
}

async fn subscribe_raw_err(rt: &RealtimeClient, channel: &str) -> Error {
    match rt.subscribe_raw(channel).await {
        Err(err) => err,
        Ok(_) => panic!("subscribe_raw({channel}) must fail"),
    }
}

#[tokio::test]
async fn l2_token_headers_then_stalled_body_times_out_via_subscribe_raw() {
    let stall = Duration::from_secs(30);
    let timeout = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::HeadersThenStall {
                status: 200,
                headers: vec![("Transfer-Encoding".into(), "chunked".into())],
                stall,
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, timeout);
    let started = Instant::now();
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    assert!(
        started.elapsed() < timeout + Duration::from_millis(800),
        "elapsed {:?} exceeded deadline+slack; body likely outside timeout",
        started.elapsed()
    );
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("timed out") || msg.contains("timeout"),
        "unexpected error: {err}"
    );
}

#[tokio::test]
async fn l2_token_no_headers_times_out_via_subscribe_raw() {
    let timeout = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::NeverRespond {
                stall: Duration::from_secs(30),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, timeout);
    let started = Instant::now();
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    assert!(started.elapsed() < timeout + Duration::from_millis(800));
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("timed out") || msg.contains("timeout"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_token_slow_drip_obeys_one_overall_deadline() {
    let timeout = Duration::from_millis(180);
    let chunks = br#"{"token":"connection-ok"}"#.iter().map(|byte| vec![*byte]).collect::<Vec<_>>();
    let http = MockHttpServer::spawn(move |req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::SlowDrip {
                status: 200,
                headers: vec![("Content-Type".into(), "application/json".into())],
                chunks: chunks.clone(),
                inter_chunk_delay: Duration::from_millis(30),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, timeout);
    let started = Instant::now();
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    assert!(
        started.elapsed() < timeout + Duration::from_millis(800),
        "slow-drip token body escaped the overall deadline"
    );
    let message = err.to_string().to_ascii_lowercase();
    assert!(
        message.contains("timeout") || message.contains("timed out"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_token_content_length_65537_rejected_via_subscribe_raw() {
    let body = vec![b'x'; 65_537];
    let http = MockHttpServer::spawn(move |req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::Raw {
                status: 200,
                headers: vec![
                    ("Content-Type".into(), "application/json".into()),
                    ("Content-Length".into(), body.len().to_string()),
                ],
                body: body.clone(),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, Duration::from_secs(2));
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("exceed") || msg.contains("too large") || msg.contains("64"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_token_chunked_oversized_rejected_via_subscribe_raw() {
    let http = MockHttpServer::spawn(|req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::ChunkedBody {
                status: 200,
                total_bytes: 70_000,
                chunk_size: 4096,
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, Duration::from_secs(2));
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("exceed") || msg.contains("too large") || msg.contains("64"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_token_empty_token_rejected_via_subscribe_raw() {
    let http = MockHttpServer::spawn(|req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::Json {
                status: 200,
                body: br#"{"token":""}"#.to_vec(),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, Duration::from_secs(2));
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("missing token") || msg.contains("token"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_token_malformed_json_rejected_via_subscribe_raw() {
    let http = MockHttpServer::spawn(|req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::Raw {
                status: 200,
                headers: vec![("Content-Type".into(), "application/json".into())],
                body: b"{not-json".to_vec(),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws = MockWsServer::spawn_hang_after_accept().await;
    let rt = private_rt(&ws, &http, Duration::from_secs(2));
    let err = subscribe_raw_err(&rt, PRIVATE_CHANNEL).await;
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("invalid token") || msg.contains("json") || msg.contains("parse"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_token_http_403_maps_to_structured_permission_denied() {
    let http = MockHttpServer::spawn(|req| {
        if req.path == "/v1/rt/token" {
            hardening_support::HttpScript::Json {
                status: 200,
                body: br#"{"token":"connection-ok"}"#.to_vec(),
            }
        } else if req.path.starts_with("/v1/rt/subscribe") {
            hardening_support::HttpScript::Json {
                status: 403,
                body: br#"{"code":"permission_denied","message":"missing transfer:read"}"#.to_vec(),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_public(active).await;
    let rt = private_rt(&ws, &http, Duration::from_secs(2));
    let err = subscribe_raw_err(&rt, "private:auth:transfers:acct:proto").await;
    match err {
        Error::PermissionDenied {
            message,
            status,
            code,
            context,
            endpoint,
        } => {
            assert_eq!(message, "missing transfer:read");
            assert_eq!(status, 403);
            assert_eq!(code, "permission_denied");
            assert!(context.contains("private:auth:transfers:acct:proto"));
            assert!(endpoint.contains("/v1/rt/subscribe?channel="));
        }
        other => panic!("expected structured permission error, got {other:?}"),
    }
}

#[tokio::test]
async fn l2_jsonrpc_headers_then_stalled_body_times_out() {
    let stall = Duration::from_secs(30);
    let timeout = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |_| hardening_support::HttpScript::HeadersThenStall {
        status: 200,
        headers: vec![
            ("Content-Type".into(), "application/json".into()),
            ("Transfer-Encoding".into(), "chunked".into()),
        ],
        stall,
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), timeout);
    let started = Instant::now();
    let err = rpc
        .request("eth_chainId", json!([]))
        .await
        .expect_err("stalled JSON-RPC body must timeout");
    assert!(started.elapsed() < timeout + Duration::from_millis(800));
    assert!(err.to_string().contains("timeout"), "{err}");
}

#[tokio::test]
async fn l2_jsonrpc_no_headers_times_out() {
    let timeout = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |_| hardening_support::HttpScript::NeverRespond {
        stall: Duration::from_secs(30),
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), timeout);
    let started = Instant::now();
    let err = rpc
        .request("eth_chainId", json!([]))
        .await
        .expect_err("no-headers JSON-RPC must timeout");
    assert!(started.elapsed() < timeout + Duration::from_millis(800));
    assert!(err.to_string().contains("timeout"), "{err}");
}

#[tokio::test]
async fn l2_jsonrpc_slow_drip_obeys_one_overall_deadline() {
    let timeout = Duration::from_millis(180);
    let chunks = br#"{"jsonrpc":"2.0","id":1,"result":"0x1"}"#
        .iter()
        .map(|byte| vec![*byte])
        .collect::<Vec<_>>();
    let http = MockHttpServer::spawn(move |_| hardening_support::HttpScript::SlowDrip {
        status: 200,
        headers: vec![("Content-Type".into(), "application/json".into())],
        chunks: chunks.clone(),
        inter_chunk_delay: Duration::from_millis(30),
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), timeout);
    let started = Instant::now();
    let err = rpc
        .request("eth_chainId", json!([]))
        .await
        .expect_err("slow-drip JSON-RPC body must time out");
    assert!(
        started.elapsed() < timeout + Duration::from_millis(800),
        "slow-drip JSON-RPC body escaped the overall deadline"
    );
    assert!(err.to_string().to_ascii_lowercase().contains("timeout"));
}

#[tokio::test]
async fn l2_jsonrpc_rejects_oversized_and_bad_envelope() {
    let http = MockHttpServer::spawn(|req| {
        if req.path.contains("big") {
            let body = vec![b'x'; 2 * 1024 * 1024];
            hardening_support::HttpScript::Raw {
                status: 200,
                headers: vec![
                    ("Content-Type".into(), "application/json".into()),
                    ("Content-Length".into(), body.len().to_string()),
                ],
                body,
            }
        } else if req.path.contains("ver") {
            hardening_support::HttpScript::Json {
                status: 200,
                body: br#"{"jsonrpc":"1.0","id":1,"result":1}"#.to_vec(),
            }
        } else if req.path.contains("noid") {
            hardening_support::HttpScript::Json {
                status: 200,
                body: br#"{"jsonrpc":"2.0","result":1}"#.to_vec(),
            }
        } else {
            hardening_support::HttpScript::Json {
                status: 200,
                body: br#"{"jsonrpc":"2.0","id":1,"result":1,"error":{"code":-1,"message":"x"}}"#
                    .to_vec(),
            }
        }
    })
    .await;
    let rpc = JsonRpcClient::new(format!("{}/ok", http.base_url()), Duration::from_secs(2));
    let err = rpc.request("eth_call", json!([])).await.unwrap_err();
    assert!(err.to_string().contains("both result and error"), "{err}");

    let big = JsonRpcClient::new(format!("{}/big", http.base_url()), Duration::from_secs(2));
    let err = big.request("eth_call", json!([])).await.unwrap_err();
    assert!(
        err.to_string().contains("exceeds") || err.to_string().contains("read body"),
        "{err}"
    );

    let ver = JsonRpcClient::new(format!("{}/ver", http.base_url()), Duration::from_secs(2));
    let err = ver.request("eth_call", json!([])).await.unwrap_err();
    assert!(
        err.to_string().to_ascii_lowercase().contains("jsonrpc") || err.to_string().contains("2.0"),
        "{err}"
    );

    let noid = JsonRpcClient::new(format!("{}/noid", http.base_url()), Duration::from_secs(2));
    let err = noid.request("eth_call", json!([])).await.unwrap_err();
    assert!(err.to_string().to_ascii_lowercase().contains("id"), "{err}");
}

#[tokio::test]
async fn l2_jsonrpc_rejects_all_remaining_invalid_envelopes() {
    let http = MockHttpServer::spawn(|req| {
        let body = if req.path.contains("missing-version") {
            br#"{"id":1,"result":1}"#.to_vec()
        } else if req.path.contains("wrong-id") {
            br#"{"jsonrpc":"2.0","id":999,"result":1}"#.to_vec()
        } else if req.path.contains("neither") {
            br#"{"jsonrpc":"2.0","id":1}"#.to_vec()
        } else {
            br#"{"jsonrpc":"2.0","id":1,"error":"not-an-object"}"#.to_vec()
        };
        hardening_support::HttpScript::Json { status: 200, body }
    })
    .await;

    for path in ["missing-version", "wrong-id", "neither", "malformed-error"] {
        let rpc = JsonRpcClient::new(
            format!("{}/{path}", http.base_url()),
            Duration::from_secs(2),
        );
        let err = rpc.request("eth_call", json!([])).await.expect_err(path);
        let message = err.to_string().to_ascii_lowercase();
        match path {
            "missing-version" => assert!(message.contains("version"), "{err}"),
            "wrong-id" => assert!(message.contains("id"), "{err}"),
            "neither" => assert!(
                message.contains("result") || message.contains("error"),
                "{err}"
            ),
            "malformed-error" => assert!(message.contains("object"), "{err}"),
            _ => unreachable!(),
        }
    }
}

#[tokio::test]
async fn l2_jsonrpc_25_concurrent_reordered_responses_succeed() {
    use std::sync::Mutex;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    // Buffer 25 requests, then reply in reverse id order (reorder-safe client).
    let buffer: Arc<Mutex<Vec<(tokio::net::TcpStream, u64)>>> = Arc::new(Mutex::new(Vec::new()));
    let buffer_h = buffer.clone();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind concurrent rpc");
    let addr = listener.local_addr().expect("addr");
    tokio::spawn(async move {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                break;
            };
            let buffer = buffer_h.clone();
            tokio::spawn(async move {
                let mut buf = vec![0u8; 64 * 1024];
                let n = match stream.read(&mut buf).await {
                    Ok(n) => n,
                    Err(_) => return,
                };
                let raw = String::from_utf8_lossy(&buf[..n]);
                let body = raw.split("\r\n\r\n").nth(1).unwrap_or("{}");
                let v: serde_json::Value = serde_json::from_str(body).unwrap_or(json!({}));
                let id = v.get("id").and_then(|i| i.as_u64()).unwrap_or(0);
                let batch = {
                    let mut guard = buffer.lock().unwrap();
                    guard.push((stream, id));
                    if guard.len() == 25 {
                        Some(std::mem::take(&mut *guard))
                    } else {
                        None
                    }
                };
                if let Some(mut batch) = batch {
                    batch.sort_by_key(|b| std::cmp::Reverse(b.1));
                    for (mut stream, id) in batch {
                        let resp = json!({"jsonrpc":"2.0","id":id,"result":format!("ok-{id}")});
                        let body = serde_json::to_vec(&resp).unwrap();
                        let head = format!(
                            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                            body.len()
                        );
                        let _ = stream.write_all(head.as_bytes()).await;
                        let _ = stream.write_all(&body).await;
                    }
                }
            });
        }
    });

    let rpc = JsonRpcClient::new(format!("http://{addr}"), Duration::from_secs(5));
    let mut tasks = Vec::new();
    for _ in 0..25 {
        let rpc = rpc.clone();
        tasks.push(tokio::spawn(async move {
            rpc.request("eth_chainId", json!([])).await
        }));
    }
    let mut ok = 0;
    for t in tasks {
        let result = t.await.expect("join").expect("rpc ok");
        assert!(result.as_str().unwrap_or("").starts_with("ok-"));
        ok += 1;
    }
    assert_eq!(ok, 25);
}

#[tokio::test]
async fn l2_jsonrpc_success_path_still_works() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Json {
        status: 200,
        body: br#"{"jsonrpc":"2.0","id":1,"result":"0x1"}"#.to_vec(),
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), Duration::from_secs(2));
    let result = rpc.request("eth_chainId", json!([])).await.expect("ok");
    assert_eq!(result, json!("0x1"));
}

#[tokio::test]
async fn l2_jsonrpc_chunked_over_1mib_rejected() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::ChunkedBody {
        status: 200,
        total_bytes: 1_100_000,
        chunk_size: 16_384,
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), Duration::from_secs(5));
    let err = rpc
        .request("eth_chainId", json!([]))
        .await
        .expect_err("chunked >1MiB must fail");
    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("exceed") || msg.contains("read body") || msg.contains("too large"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_jsonrpc_malformed_json_rejected() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Raw {
        status: 200,
        headers: vec![("Content-Type".into(), "application/json".into())],
        body: b"{broken".to_vec(),
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), Duration::from_secs(2));
    let err = rpc.request("eth_chainId", json!([])).await.unwrap_err();
    assert!(
        err.to_string().to_ascii_lowercase().contains("json"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_jsonrpc_error_object_returns_transport_error() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Json {
        status: 200,
        body: br#"{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"boom"}}"#.to_vec(),
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), Duration::from_secs(2));
    let err = rpc.request("eth_call", json!([])).await.unwrap_err();
    assert!(err.to_string().contains("boom"), "{err}");
}

#[tokio::test]
async fn l2_jsonrpc_null_result_is_preserved() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Json {
        status: 200,
        body: br#"{"jsonrpc":"2.0","id":1,"result":null}"#.to_vec(),
    })
    .await;
    let rpc = JsonRpcClient::new(http.base_url(), Duration::from_secs(2));
    let result = rpc.request("eth_call", json!([])).await.expect("null ok");
    assert!(result.is_null(), "expected null result, got {result}");
}

#[tokio::test]
async fn l2_close_aborts_subscription_promptly_against_local_ws() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_public(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let sub = rt.subscribe_raw(PUBLIC_CHANNEL).await.expect("subscribe");
    wait_until(
        || active.load(Ordering::SeqCst) >= 1,
        Duration::from_secs(2),
    )
    .await;
    let started = Instant::now();
    sub.close();
    wait_until(|| !sub.is_alive(), Duration::from_millis(750)).await;
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert!(
        started.elapsed() < Duration::from_millis(750),
        "close lingered {:?}",
        started.elapsed()
    );
}

#[tokio::test]
async fn l2_drop_idle_subscription_peer_closes_promptly() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_public(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let sub = rt.subscribe_raw(PUBLIC_CHANNEL).await.expect("subscribe");
    wait_until(
        || active.load(Ordering::SeqCst) >= 1,
        Duration::from_secs(2),
    )
    .await;
    let started = Instant::now();
    drop(sub);
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert!(started.elapsed() < Duration::from_millis(750));
}

#[tokio::test]
async fn l2_hundred_sub_close_returns_conn_count_to_baseline() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_public(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let mut subs = Vec::new();
    for _ in 0..100 {
        subs.push(rt.subscribe_raw(PUBLIC_CHANNEL).await.expect("subscribe"));
    }
    wait_until(
        || active.load(Ordering::SeqCst) >= 100,
        Duration::from_secs(5),
    )
    .await;
    let started = Instant::now();
    for sub in &subs {
        sub.close();
    }
    drop(subs);
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert!(
        started.elapsed() < Duration::from_millis(750),
        "100-sub close soak exceeded 750ms: {:?}",
        started.elapsed()
    );
}

#[tokio::test]
async fn l2_cancel_subscribe_raw_during_token_body_stall_cleans_peers() {
    let stall = Duration::from_secs(30);
    let http = MockHttpServer::spawn(move |req| {
        if req.path.starts_with("/v1/rt/") {
            hardening_support::HttpScript::HeadersThenStall {
                status: 200,
                headers: vec![("Transfer-Encoding".into(), "chunked".into())],
                stall,
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let ws_active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_hang_after_accept_counted(ws_active.clone()).await;
    let rt = private_rt(&ws, &http, Duration::from_secs(30));

    let join = tokio::spawn(async move { rt.subscribe_raw(PRIVATE_CHANNEL).await });
    // Arming wait only: under CI load (shared with the 10k auth soak) the
    // subscribe task may take longer than 2s to open the token HTTP request.
    wait_until(
        || http.in_flight.load(Ordering::SeqCst) >= 1,
        Duration::from_secs(5),
    )
    .await;
    let started = Instant::now();
    join.abort();
    let _ = join.await;
    wait_until(
        || http.in_flight.load(Ordering::SeqCst) == 0 && ws_active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert!(
        started.elapsed() < Duration::from_millis(750),
        "cancel during token stall lingered {:?}",
        started.elapsed()
    );
}

#[tokio::test]
async fn l2_cancel_subscribe_raw_during_centrifugo_wait_cleans_peers() {
    let ws_active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_hang_after_accept_counted(ws_active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let join = tokio::spawn(async move { rt.subscribe_raw(PUBLIC_CHANNEL).await });
    wait_until(
        || ws_active.load(Ordering::SeqCst) >= 1,
        Duration::from_secs(5),
    )
    .await;
    let started = Instant::now();
    join.abort();
    let _ = join.await;
    wait_until(
        || ws_active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert!(
        started.elapsed() < Duration::from_millis(750),
        "cancel during Centrifugo wait lingered {:?}",
        started.elapsed()
    );
}

#[tokio::test]
async fn l2_close_during_reconnect_backoff_no_extra_connect() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_disconnect_after_handshake(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let sub = rt.subscribe_raw(PUBLIC_CHANNEL).await.expect("subscribe");
    wait_until(
        || ws.connects.load(Ordering::SeqCst) >= 1,
        Duration::from_secs(2),
    )
    .await;
    // After forced disconnect, the client enters its jittered reconnect backoff.
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_secs(2),
    )
    .await;
    let connects_before = ws.connects.load(Ordering::SeqCst);
    sub.close();
    tokio::time::sleep(Duration::from_millis(1200)).await;
    let connects_after = ws.connects.load(Ordering::SeqCst);
    assert_eq!(
        connects_before, connects_after,
        "close during reconnect backoff must not start an extra connect ({connects_before} -> {connects_after})"
    );
    assert!(!sub.is_alive());
}

#[tokio::test]
async fn l2_realtime_terminal_error_is_callback_and_result_observable() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_disconnect_after_handshake(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let mut sub = rt
        .subscribe_proto_with_options(PUBLIC_CHANNEL, |bytes| Ok(bytes.to_vec()), false)
        .await
        .expect("initial handshake");

    let (error_tx, mut error_rx) = tokio::sync::mpsc::unbounded_channel();
    sub.set_on_error(move |err| {
        let _ = error_tx.send(err);
    });

    let callback_error = tokio::time::timeout(Duration::from_millis(750), error_rx.recv())
        .await
        .expect("error callback must be prompt")
        .expect("error callback channel");
    let callback_message = callback_error.to_string().to_ascii_lowercase();
    assert!(
        callback_message.contains("closed")
            || callback_message.contains("reset")
            || callback_message.contains("eof"),
        "unexpected callback error: {callback_error}"
    );

    let result_error = tokio::time::timeout(Duration::from_millis(750), sub.recv_result())
        .await
        .expect("recv_result must unblock")
        .expect_err("terminal feed failure must not look like clean EOF");
    let result_message = result_error.to_string().to_ascii_lowercase();
    assert!(
        result_message.contains("closed")
            || result_message.contains("reset")
            || result_message.contains("eof"),
        "unexpected recv_result error: {result_error}"
    );
    assert!(!sub.is_alive());
    sub.close();
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
}

#[tokio::test]
async fn l2_realtime_oversized_binary_message_fails_closed() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_oversized_after_handshake(
        active.clone(),
        MAX_REALTIME_MESSAGE_BYTES + 1,
    )
    .await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let sub = rt
        .subscribe_proto_with_options(PUBLIC_CHANNEL, |bytes| Ok(bytes.to_vec()), false)
        .await
        .expect("initial handshake");
    wait_until(
        || sub.err().is_some() && !sub.is_alive(),
        Duration::from_secs(3),
    )
    .await;
    let error = sub.err().expect("oversized message error").to_string();
    let error = error.to_ascii_lowercase();
    assert!(
        error.contains("size")
            || error.contains("too long")
            || error.contains("capacity")
            || error.contains("space limit"),
        "unexpected oversized-message error: {error}"
    );
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
}

#[tokio::test]
async fn l2_close_during_reconnect_snapshot_retry_cancels_fetch_and_socket() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_disconnect_once_then_idle(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let attempts = Arc::new(AtomicUsize::new(0));
    let fetch_attempts = attempts.clone();
    let (stalled_tx, stalled_rx) = tokio::sync::oneshot::channel::<()>();
    let stalled_rx = Arc::new(std::sync::Mutex::new(Some(stalled_rx)));
    let fetch_stalled_rx = stalled_rx.clone();
    let sts = SnapshotThenStream::new(SnapshotThenStreamConfig {
        client: rt,
        channel: PUBLIC_CHANNEL.into(),
        decode: Arc::new(|_bytes| Ok(1u8)),
        fetch_snapshot: Arc::new(move || {
            let attempt = fetch_attempts.fetch_add(1, Ordering::SeqCst);
            if attempt == 0 {
                return async { Ok("initial".to_owned()) }.boxed();
            }
            if attempt == 1 {
                return async { Err(Error::transport("retry once")) }.boxed();
            }
            let stalled = fetch_stalled_rx
                .lock()
                .expect("stalled snapshot lock")
                .take()
                .expect("only one stalled retry");
            async move {
                let _ = stalled.await;
                Ok("late snapshot".to_owned())
            }
            .boxed()
        }),
        read_publication: Arc::new(|p| vec![p]),
        apply_snapshot: Arc::new(|_snapshot, _pending| {}),
        apply_live_publications: Arc::new(|_publications| {}),
        max_buffered: 8,
        on_reconnect: None,
        on_snapshot_refresh: None,
        on_error: None,
    });

    sts.start().await.expect("initial snapshot");
    wait_until(
        || attempts.load(Ordering::SeqCst) >= 3 && active.load(Ordering::SeqCst) == 1,
        Duration::from_secs(5),
    )
    .await;
    let started = Instant::now();
    sts.close();
    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert!(
        stalled_tx.send(()).is_err(),
        "closing during the second refresh attempt must drop the stalled fetch future"
    );
    assert!(
        started.elapsed() < Duration::from_millis(750),
        "combined retry/cancellation cleanup lingered {:?}",
        started.elapsed()
    );
}

#[tokio::test]
async fn l2_scale_format_and_catalog_reject_panic_boundary() {
    assert!(format_qty_scaled(1, 18).is_ok());
    assert!(format_qty_scaled(1, MAX_PROTOCOL_SCALE).is_ok());
    for scale in [37u32, 65534, 65535, 65536, u32::MAX] {
        assert!(
            format_qty_scaled(1, scale).is_err(),
            "scale {scale} must err"
        );
        assert!(format_ledger_u128("1", scale).is_err());
        let formatted = Quantity::from_scaled(1, Some(8), QuantityDomain::OrderBase, None, None)
            .unwrap()
            .format(Some(scale));
        assert!(formatted.is_err(), "Quantity::format({scale}) must err");
        let construct =
            Quantity::from_scaled(1, Some(scale), QuantityDomain::OrderBase, None, None);
        assert!(
            construct.is_err(),
            "from_scaled with scale {scale} must err at boundary"
        );
        let panic = std::panic::catch_unwind(|| {
            let _ = format_qty_scaled(1, scale);
        });
        assert!(panic.is_ok(), "format must not panic at scale {scale}");
    }

    let catalogs = polyester::catalogs::Manager::new();
    let err = catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "BTC-USDT",
                "symbol_id": 1,
                "base_quantity_scale": 65535
            }]
        }))
        .expect_err("catalog must reject panic-boundary scale");
    assert!(err.to_string().contains("scale"));
    assert_eq!(catalogs.base_quantity_scale_for_symbol("BTC-USDT"), None);

    let err = catalogs
        .hydrate_spot_config_json(json!({
            "pairs": [{
                "symbol": "ETH-USDT",
                "symbol_id": 4294967296u64,
                "base_quantity_scale": 8
            }]
        }))
        .expect_err("catalog must reject symbol_id above u32");
    assert!(err.to_string().contains("symbol_id") || err.to_string().contains("u32"));
}

#[tokio::test]
async fn l2_wait_for_catalogs_fail_closed_on_http_500() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Json {
        status: 500,
        body: br#"{"error":"nope"}"#.to_vec(),
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_millis(500),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("HTTP 500 must fail closed");
    assert!(
        err.to_string().contains("catalog hydration failed"),
        "{err}"
    );
    assert!(client.catalogs_last_error().is_some());
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        None
    );
}

#[tokio::test]
async fn l2_wait_for_catalogs_fail_closed_on_empty_body() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Raw {
        status: 200,
        headers: vec![("Content-Type".into(), "application/json".into())],
        body: Vec::new(),
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_millis(500),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("empty body must fail closed");
    assert!(err.to_string().contains("catalog"), "{err}");
    assert!(client.catalogs_last_error().is_some());
}

#[tokio::test]
async fn l2_wait_for_catalogs_fail_closed_on_malformed_protobuf() {
    let http = MockHttpServer::spawn(|req| {
        if req.path == SPOT_CONFIG_PATH {
            hardening_support::HttpScript::Raw {
                status: 200,
                headers: vec![
                    ("Content-Type".into(), "application/proto".into()),
                    ("Content-Length".into(), "1".into()),
                ],
                // Field zero with an unsupported wire type cannot decode as a
                // protobuf GetSpotConfigResponse.
                body: vec![0x0f],
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_millis(500),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("malformed protobuf must fail closed");
    assert!(!client.catalogs.is_ready(), "{err}");
    assert!(client.catalogs_last_error().is_some());
}

#[tokio::test]
async fn l2_wait_for_catalogs_fail_closed_on_malformed_config() {
    let http = MockHttpServer::spawn(|_| hardening_support::HttpScript::Json {
        status: 200,
        body: br#"{"pairs":[{"symbol":"BTC-USDT","symbol_id":1,"base_quantity_scale":65535}]}"#
            .to_vec(),
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_millis(500),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("bad scale in config must fail closed");
    assert!(
        err.to_string().contains("scale") || err.to_string().contains("catalog"),
        "{err}"
    );
}

#[tokio::test]
async fn l2_wait_for_catalogs_rejects_oversized_content_length_response() {
    let body = vec![0u8; MAX_CONNECT_RESPONSE_BYTES + 1];
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            hardening_support::HttpScript::Raw {
                status: 200,
                headers: vec![
                    ("Content-Type".into(), "application/proto".into()),
                    ("Content-Length".into(), body.len().to_string()),
                ],
                body: body.clone(),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_secs(2),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");

    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("oversized catalog response must fail closed");
    let message = err.to_string().to_ascii_lowercase();
    assert!(
        message.contains("size")
            || message.contains("limit")
            || message.contains("resource exhausted")
            || message.contains("catalog"),
        "{err}"
    );
    assert!(!client.catalogs.is_ready());
    assert!(client.catalogs_last_error().is_some());
}

#[tokio::test]
async fn l2_wait_for_catalogs_rejects_oversized_chunked_response() {
    let http = MockHttpServer::spawn(|req| {
        if req.path == SPOT_CONFIG_PATH {
            hardening_support::HttpScript::ChunkedBody {
                status: 200,
                total_bytes: MAX_CONNECT_RESPONSE_BYTES + 1,
                chunk_size: 64 * 1024,
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_secs(2),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");

    client
        .wait_for_catalogs()
        .await
        .expect_err("oversized chunked catalog response must fail closed");
    assert!(!client.catalogs.is_ready());
    assert!(client.catalogs_last_error().is_some());
}

#[tokio::test]
async fn l2_concurrent_wait_for_catalogs_share_one_attempt() {
    let stall = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |_| hardening_support::HttpScript::HeadersThenStall {
        status: 500,
        headers: vec![("Content-Type".into(), "application/json".into())],
        stall,
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_millis(200),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");

    let (r1, r2) = tokio::join!(client.wait_for_catalogs(), client.wait_for_catalogs());
    assert!(r1.is_err());
    assert!(r2.is_err());
    // Spot + zipper = 2 requests for one shared attempt (not 4).
    let n = http.requests.load(Ordering::SeqCst);
    assert!(
        n <= 2,
        "concurrent waiters must share one attempt; saw {n} HTTP requests"
    );
}

#[tokio::test]
async fn l2_snapshot_then_stream_start_survives_immediate_disconnect_repeatedly() {
    const ATTEMPTS: usize = 60;
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_disconnect_after_handshake(active.clone()).await;

    for attempt in 1..=ATTEMPTS {
        let rt =
            RealtimeClient::with_timeout(ws.ws_url(), "", None, None, Duration::from_millis(500));
        let sts = SnapshotThenStream::new(SnapshotThenStreamConfig {
            client: rt,
            channel: PUBLIC_CHANNEL.into(),
            decode: Arc::new(|_b| Ok(1u8)),
            fetch_snapshot: Arc::new(|| async { Ok("initial".to_string()) }.boxed()),
            read_publication: Arc::new(|p| vec![p]),
            apply_snapshot: Arc::new(|_s, _p| {}),
            apply_live_publications: Arc::new(|_p| {}),
            max_buffered: 8,
            on_reconnect: None,
            on_snapshot_refresh: None,
            on_error: None,
        });

        let started = Instant::now();
        tokio::time::timeout(Duration::from_millis(750), sts.start())
            .await
            .unwrap_or_else(|_| panic!("attempt {attempt}/{ATTEMPTS}: start hung"))
            .unwrap_or_else(|err| {
                panic!("attempt {attempt}/{ATTEMPTS}: valid handshake failed: {err}")
            });
        assert!(
            started.elapsed() < Duration::from_millis(750),
            "attempt {attempt}/{ATTEMPTS}: start exceeded its bound"
        );
        sts.close();
    }

    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert_eq!(
        active.load(Ordering::SeqCst),
        0,
        "repeated immediate disconnects leaked sockets"
    );
}

#[tokio::test]
async fn l2_snapshot_then_stream_start_obeys_configured_deadline() {
    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_hang_after_accept_counted(active.clone()).await;
    let configured_timeout = Duration::from_millis(150);
    let rt = RealtimeClient::with_timeout(ws.ws_url(), "", None, None, configured_timeout);
    let sts = SnapshotThenStream::new(SnapshotThenStreamConfig {
        client: rt,
        channel: PUBLIC_CHANNEL.into(),
        decode: Arc::new(|_b| Ok(1u8)),
        fetch_snapshot: Arc::new(|| async { Ok("unreachable".to_string()) }.boxed()),
        read_publication: Arc::new(|p| vec![p]),
        apply_snapshot: Arc::new(|_s, _p| {}),
        apply_live_publications: Arc::new(|_p| {}),
        max_buffered: 8,
        on_reconnect: None,
        on_snapshot_refresh: None,
        on_error: None,
    });

    let started = Instant::now();
    let err = tokio::time::timeout(Duration::from_millis(750), sts.start())
        .await
        .expect("configured startup deadline must terminate")
        .expect_err("stalled Centrifugo handshake must fail");
    assert!(
        started.elapsed() >= configured_timeout,
        "startup returned before its configured deadline"
    );
    assert!(
        started.elapsed() < Duration::from_millis(750),
        "startup exceeded its configured deadline plus test slack"
    );
    assert!(
        err.to_string().to_ascii_lowercase().contains("timed out"),
        "unexpected startup error: {err}"
    );
    assert!(sts.is_disposed());
    assert!(sts.err().is_some());

    wait_until(
        || active.load(Ordering::SeqCst) == 0,
        Duration::from_millis(750),
    )
    .await;
    assert_eq!(
        active.load(Ordering::SeqCst),
        0,
        "startup timeout leaked the stalled websocket"
    );
}

#[tokio::test]
async fn l2_snapshot_then_stream_reconnect_snapshot_fail_sets_err_and_not_ready() {
    use std::sync::atomic::AtomicBool;

    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_disconnect_after_handshake(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let attempts = Arc::new(AtomicUsize::new(0));
    let fetch_attempts = attempts.clone();
    // Succeed until the initial start marks ready; fail on reconnect refresh.
    let fail_after_ready = Arc::new(AtomicBool::new(false));
    let fail_flag = fail_after_ready.clone();
    let reported_errors = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
    let reported_errors_cb = reported_errors.clone();
    let sts = SnapshotThenStream::new(SnapshotThenStreamConfig {
        client: rt,
        channel: PUBLIC_CHANNEL.into(),
        decode: Arc::new(|_b| Ok(1u8)),
        fetch_snapshot: Arc::new(move || {
            fetch_attempts.fetch_add(1, Ordering::SeqCst);
            let fail = fail_flag.load(Ordering::SeqCst);
            async move {
                if fail {
                    Err(Error::transport("snapshot refresh failed"))
                } else {
                    Ok("initial".to_string())
                }
            }
            .boxed()
        }),
        read_publication: Arc::new(|p| vec![p]),
        apply_snapshot: Arc::new(|_s, _p| {}),
        apply_live_publications: Arc::new(|_p| {}),
        max_buffered: 8,
        on_reconnect: None,
        on_snapshot_refresh: None,
        on_error: Some(Arc::new(move |err| {
            reported_errors_cb
                .lock()
                .expect("reported errors")
                .push(err.to_string());
        })),
    });
    sts.start().await.expect("initial start");
    assert!(sts.is_ready());
    fail_after_ready.store(true, Ordering::SeqCst);
    // Wait for disconnect → reconnect → failing snapshot (with one retry) → fail-closed.
    wait_until(
        || sts.err().is_some() && !sts.is_ready(),
        Duration::from_secs(5),
    )
    .await;
    assert!(sts.err().is_some());
    assert!(!sts.is_ready());
    assert!(
        sts.is_disposed(),
        "exhausted snapshot recovery must terminate fail-closed"
    );
    let reported_errors = reported_errors.lock().expect("reported errors");
    assert!(
        reported_errors
            .iter()
            .any(|err| err.contains("snapshot refresh failed")),
        "snapshot refresh failure was not surfaced to on_error: {reported_errors:?}"
    );
    assert!(
        attempts.load(Ordering::SeqCst) >= 2,
        "expected initial success + reconnect retries"
    );
    sts.close();
}

#[tokio::test]
async fn l2_balances_list_connect_response_preserves_ledger_scaled_integer() {
    use polyester::codecs::format_ledger_u128;
    use polyester::proto::ledger::read::v1::GetBalancesRequest;
    use polyester::proto::ledger::read::v1::{
        AssetBalance as ProtoAssetBalance, GetBalancesResponse,
    };
    use polyester::proto::polyester::r#type::v1::U128;

    const ONE_POINT_FIVE_E18: u64 = 1_500_000_000_000_000_000;
    let fixture = GetBalancesResponse {
        balances: vec![ProtoAssetBalance {
            asset_id: 1,
            trading: U128 {
                hi: 0,
                lo: ONE_POINT_FIVE_E18,
                ..Default::default()
            }
            .into(),
            funding: U128 {
                hi: 0,
                lo: ONE_POINT_FIVE_E18,
                ..Default::default()
            }
            .into(),
            ..Default::default()
        }],
        ..Default::default()
    };
    let http = MockHttpServer::spawn(move |req| {
        if req.path == GET_BALANCES_PATH {
            connect_proto_ok(&fixture)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        timeout: Duration::from_secs(2),
        ..Default::default()
    })
    .expect("client");
    let list = client
        .balances
        .list(GetBalancesRequest::default())
        .await
        .expect("balances.list");
    assert_eq!(list.balances.len(), 1);
    assert_eq!(list.balances[0].trading, "1500000000000000000");
    assert_eq!(list.balances[0].funding, "1500000000000000000");
    assert_eq!(
        format_ledger_u128(&list.balances[0].trading, 18).expect("format"),
        "1.5"
    );
}

#[tokio::test]
async fn l2_wait_for_order_trades_complete_polls_get_order_until_trade_sum_matches() {
    use polyester::proto::orders::v1::{GetOrderResponse, Order, OrderStatus, UserTrade};

    let calls = Arc::new(AtomicUsize::new(0));
    let calls_h = calls.clone();
    let http = MockHttpServer::spawn(move |req| {
        if req.path != GET_ORDER_PATH {
            return hardening_support::HttpScript::NotFound;
        }
        let n = calls_h.fetch_add(1, Ordering::SeqCst) + 1;
        let order = Order {
            order_id: 1,
            symbol_id: 1,
            status: OrderStatus::Filled.into(),
            cum_qty_scaled: 100,
            ..Default::default()
        };
        let resp = if n == 1 {
            GetOrderResponse {
                order: Some(order).into(),
                trades: vec![],
                ..Default::default()
            }
        } else {
            GetOrderResponse {
                order: Some(order).into(),
                trades: vec![
                    UserTrade {
                        symbol_id: 1,
                        qty_scaled: 40,
                        fee_scaled: 1,
                        fee_asset: polyester::proto::orders::v1::FeeAsset::Base.into(),
                        referral_share_scaled: 1,
                        ..Default::default()
                    },
                    UserTrade {
                        symbol_id: 1,
                        qty_scaled: 60,
                        ..Default::default()
                    },
                ],
                ..Default::default()
            }
        };
        connect_proto_ok(&resp)
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        timeout: Duration::from_secs(2),
        ..Default::default()
    })
    .expect("client");
    let result = client
        .orders
        .wait_for_order_trades_complete(
            polyester::models::OrderKey::OrderId("1".into()),
            Duration::from_secs(2),
        )
        .await
        .expect("wait complete");
    assert!(calls.load(Ordering::SeqCst) >= 2);
    let cum = result
        .order
        .as_ref()
        .and_then(|o| o.cum_qty.as_ref())
        .map(|q| q.as_scaled())
        .expect("cum");
    let sum: i64 = result
        .trades
        .iter()
        .map(|t| t.qty.as_ref().map(|q| q.as_scaled()).unwrap_or(0))
        .sum();
    assert_eq!(cum, 100);
    assert_eq!(sum, 100);
    assert_eq!(result.trades[0].fee_asset, "base");
    assert_eq!(result.trades[0].fee_scaled, "1");
    assert_eq!(result.trades[0].referral_share_scaled, "1");
}

#[tokio::test]
async fn l2_wait_for_order_trades_complete_enforces_overall_deadline() {
    let http = MockHttpServer::spawn(|req| {
        if req.path == GET_ORDER_PATH {
            hardening_support::HttpScript::HeadersThenStall {
                status: 200,
                headers: vec![
                    ("Content-Type".into(), "application/proto".into()),
                    ("Transfer-Encoding".into(), "chunked".into()),
                ],
                stall: Duration::from_secs(30),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        timeout: Duration::from_secs(5),
        ..Default::default()
    })
    .expect("client");
    let helper_timeout = Duration::from_millis(250);
    let started = Instant::now();
    let err = client
        .orders
        .wait_for_order_trades_complete(
            polyester::models::OrderKey::OrderId("1".into()),
            helper_timeout,
        )
        .await
        .expect_err("helper deadline must cover the in-flight GetOrder call");
    assert!(
        started.elapsed() < Duration::from_secs(1),
        "helper exceeded overall deadline: {:?}",
        started.elapsed()
    );
    assert!(err.to_string().contains("timed out"), "{err}");
}

fn spot_config_fixture() -> polyester::proto::marketdata::v1::GetSpotConfigResponse {
    use polyester::proto::marketdata::v1::{GetSpotConfigResponse, PairConfig};
    GetSpotConfigResponse {
        pairs: vec![PairConfig {
            symbol_id: 1,
            symbol: "BTC-USDT".into(),
            base_asset: "BTC".into(),
            quote_asset: "USDT".into(),
            base_quantity_scale: 8,
            ..Default::default()
        }],
        ts_sec: 1,
        ..Default::default()
    }
}

fn zipper_config_fixture() -> polyester::proto::chain::zipper::v1::GetDepositWithdrawConfigResponse
{
    use polyester::proto::chain::zipper::v1::{AssetConfig, GetDepositWithdrawConfigResponse};
    GetDepositWithdrawConfigResponse {
        polyester_chain_id: 1,
        ts_sec: 1,
        assets: vec![AssetConfig {
            asset: "USDT".into(),
            ledger_id: 99,
            quantity_scale: 6,
            ..Default::default()
        }],
        ..Default::default()
    }
}

#[tokio::test]
async fn l2_wait_for_catalogs_hydrates_spot_and_zipper_then_formats() {
    use polyester::proto::marketdata::v1::GetTradesResponse;
    use polyester::{Quantity, QuantityDomain};

    let spot = spot_config_fixture();
    let zipper = zipper_config_fixture();
    let trades = GetTradesResponse::default();
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            connect_proto_ok(&spot)
        } else if req.path == ZIPPER_CONFIG_PATH {
            connect_proto_ok(&zipper)
        } else if req.path == GET_TRADES_PATH {
            connect_proto_ok(&trades)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_secs(2),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    client.wait_for_catalogs().await.expect("hydrate");
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        Some(8)
    );
    assert_eq!(client.catalogs.symbol_id_for_symbol("BTC-USDT"), Some(1));
    assert_eq!(client.catalogs.ledger_id_for_asset("USDT"), Some(99));
    let qty = Quantity::from_scaled(100_000_000, Some(8), QuantityDomain::OrderBase, None, None)
        .expect("qty");
    assert_eq!(qty.format(None).expect("format"), "1");
    // Public symbol resolution path after hydrate.
    let _ = client
        .market_data
        .get_trades("BTC-USDT", Some(1))
        .await
        .expect("get_trades resolves hydrated symbol");
}

#[tokio::test]
async fn l2_wait_for_catalogs_no_headers_stalls_then_times_out() {
    let timeout = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |_| hardening_support::HttpScript::NeverRespond {
        stall: Duration::from_secs(30),
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout,
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let started = Instant::now();
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("no-headers must fail");
    assert!(started.elapsed() < timeout + Duration::from_millis(1200));
    assert!(err.to_string().contains("catalog"), "{err}");
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        None
    );
}

#[tokio::test]
async fn l2_wait_for_catalogs_headers_then_body_stall_times_out() {
    let stall = Duration::from_secs(30);
    let timeout = Duration::from_millis(400);
    let http = MockHttpServer::spawn(move |_| hardening_support::HttpScript::HeadersThenStall {
        status: 200,
        headers: vec![
            ("Content-Type".into(), "application/proto".into()),
            ("Transfer-Encoding".into(), "chunked".into()),
        ],
        stall,
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout,
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let started = Instant::now();
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("body stall must fail");
    assert!(started.elapsed() < timeout + Duration::from_millis(1200));
    assert!(err.to_string().contains("catalog"), "{err}");
}

#[tokio::test]
async fn l2_batch_cancel_rejects_inconsistent_server_counts_through_public_service() {
    use polyester::models::BatchCancelItem;
    use polyester::proto::orders::v1::{
        BatchCancelOrdersResponse, BatchCancelResultItem as ProtoItem,
    };

    let response = BatchCancelOrdersResponse {
        results: vec![ProtoItem {
            status: "accepted".into(),
            order_id: 9,
            ..Default::default()
        }],
        accepted_count: 0,
        rejected_count: 1,
        ..Default::default()
    };
    let http = MockHttpServer::spawn(move |req| {
        if req.path == "/orders.v1.OrdersService/BatchCancelOrders" {
            connect_proto_ok(&response)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .expect("client");

    let err = client
        .orders
        .batch_cancel(
            vec![BatchCancelItem {
                key: polyester::models::OrderKey::OrderId("9".into()),
                symbol_id: None,
            }],
            None,
            Some("count-mismatch".into()),
        )
        .await
        .expect_err("service must reject an ambiguous batch result");
    assert!(err.to_string().contains("response counts"), "{err}");
}

#[tokio::test]
async fn l2_cancel_all_rejects_empty_and_unknown_status_through_public_service() {
    use polyester::proto::orders::v1::CancelAllOrdersResponse;

    for status in ["", "ok", "maybe"] {
        let response = CancelAllOrdersResponse {
            status: status.into(),
            matched_orders: 1,
            submitted_cancels: 1,
            ..Default::default()
        };
        let http = MockHttpServer::spawn(move |req| {
            if req.path == "/orders.v1.OrdersService/CancelAllOrders" {
                connect_proto_ok(&response)
            } else {
                hardening_support::HttpScript::NotFound
            }
        })
        .await;
        let client = Client::new(Config {
            api_url: http.base_url(),
            api_key_id: Some("ak_test".into()),
            api_private_key: Some(TEST_KEY.into()),
            hydrate_catalogs: false,
            ..Default::default()
        })
        .expect("client");

        let err = client
            .orders
            .cancel_all(None, false, None)
            .await
            .expect_err("empty/unknown cancel-all status must fail closed");
        assert!(err.to_string().contains("unknown status"), "{err}");
    }
}

#[tokio::test]
async fn l2_cancel_all_after_rejects_empty_and_unknown_status_through_public_service() {
    use polyester::proto::orders::v1::CancelAllAfterResponse;

    for status in ["", "submitted", "maybe"] {
        let response = CancelAllAfterResponse {
            status: status.into(),
            effective_timeout_sec: 15,
            ..Default::default()
        };
        let http = MockHttpServer::spawn(move |req| {
            if req.path == "/orders.v1.OrdersService/CancelAllAfter" {
                connect_proto_ok(&response)
            } else {
                hardening_support::HttpScript::NotFound
            }
        })
        .await;
        let client = Client::new(Config {
            api_url: http.base_url(),
            api_key_id: Some("ak_test".into()),
            api_private_key: Some(TEST_KEY.into()),
            hydrate_catalogs: false,
            ..Default::default()
        })
        .expect("client");

        let err = client
            .orders
            .cancel_all_after(15, None, None, None)
            .await
            .expect_err("empty/unknown cancel-all-after status must fail closed");
        assert!(err.to_string().contains("unknown status"), "{err}");
    }
}

#[tokio::test]
async fn l2_batch_replace_rejects_inconsistent_server_counts_through_public_service() {
    use polyester::Price;
    use polyester::models::BatchReplaceItem;
    use polyester::proto::orders::v1::{
        BatchReplaceAdmissionItem as ProtoItem, BatchReplaceAdmissionStatus,
        BatchReplaceItemAdmissionStatus, BatchReplaceOrdersResponse,
    };

    let spot = spot_config_fixture();
    let zipper = zipper_config_fixture();
    let response = BatchReplaceOrdersResponse {
        batch_request_id: 9,
        status: BatchReplaceAdmissionStatus::Admitted.into(),
        results: vec![ProtoItem {
            status: BatchReplaceItemAdmissionStatus::Admitted.into(),
            replacement_order_id: 9,
            ..Default::default()
        }],
        accepted_count: 0,
        rejected_count: 1,
        ..Default::default()
    };
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            connect_proto_ok(&spot)
        } else if req.path == ZIPPER_CONFIG_PATH {
            connect_proto_ok(&zipper)
        } else if req.path == "/orders.v1.OrdersService/BatchReplaceOrders" {
            connect_proto_ok(&response)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    client.wait_for_catalogs().await.expect("catalogs");

    let err = client
        .orders
        .batch_replace(
            vec![BatchReplaceItem {
                key: polyester::models::OrderKey::OrderId("9".into()),
                new_price: Some(Price::from_ticks(1, Some("BTC-USDT".into())).expect("price")),
                new_qty: None,
                new_attached_risk: None,
                new_client_order_id: None,
            }],
            "BTC-USDT",
            None,
            Some("replace-count-mismatch".into()),
        )
        .await
        .expect_err("service must reject an ambiguous batch result");
    assert!(err.to_string().contains("response counts"), "{err}");
}

#[tokio::test]
async fn l2_columnar_candles_reject_misaligned_columns_through_public_service() {
    use polyester::models::GetCandlesOpts;
    use polyester::proto::marketdata::v1::{GetCandlesColumnsResponse, Timeframe};

    let spot = spot_config_fixture();
    let zipper = zipper_config_fixture();
    let response = GetCandlesColumnsResponse {
        symbol_id: 1,
        timeframe: Timeframe::Min1.into(),
        ts_sec: vec![10, 20],
        open: vec![1, 2],
        high: vec![1],
        low: vec![1, 2],
        close: vec![1, 2],
        volume: vec![1, 2],
        ..Default::default()
    };
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            connect_proto_ok(&spot)
        } else if req.path == ZIPPER_CONFIG_PATH {
            connect_proto_ok(&zipper)
        } else if req.path == "/marketdata.v1.MarketDataService/GetCandlesColumns" {
            connect_proto_ok(&response)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    client.wait_for_catalogs().await.expect("catalogs");

    let err = client
        .market_data
        .get_candles_columns(GetCandlesOpts {
            symbol: Some("BTC-USDT".into()),
            timeframe: "1m".into(),
            ..Default::default()
        })
        .await
        .expect_err("service must reject misaligned OHLCV columns");
    assert!(err.to_string().contains("response lengths"), "{err}");
}

#[tokio::test]
async fn l2_lifecycle_get_rejects_missing_required_flow_through_public_service() {
    use polyester::proto::chain::lifecycle::v1::GetFlowResponse;

    let response = GetFlowResponse::default();
    let http = MockHttpServer::spawn(move |req| {
        if req.path == "/chain.lifecycle.v1.LifecycleReadService/GetFlowById" {
            connect_proto_ok(&response)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .expect("client");

    let err = client
        .lifecycle
        .get_flow("flow-1")
        .await
        .expect_err("service must reject a missing required flow");
    assert!(err.to_string().contains("missing flow"), "{err}");
}

#[tokio::test]
async fn l2_create_deposit_address_rejects_missing_entity_through_public_service() {
    use polyester::proto::chain::deposit::v1::{
        CreateDepositAddressRequest, CreateDepositAddressResponse,
    };

    let response = CreateDepositAddressResponse::default();
    let http = MockHttpServer::spawn(move |req| {
        if req.path == "/chain.deposit.v1.DepositAddressService/CreateDepositAddress" {
            connect_proto_ok(&response)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        api_key_id: Some("ak_test".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .expect("client");

    let err = client
        .deposit
        .create_address(CreateDepositAddressRequest {
            chain_id: 1,
            ..Default::default()
        })
        .await
        .expect_err("service must reject a missing deposit address");
    assert!(err.to_string().contains("missing deposit_address"), "{err}");
}

#[tokio::test]
async fn l2_wait_for_catalogs_success_path() {
    let spot = spot_config_fixture();
    let zipper = zipper_config_fixture();
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            connect_proto_ok(&spot)
        } else if req.path == ZIPPER_CONFIG_PATH {
            connect_proto_ok(&zipper)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_secs(2),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    client.wait_for_catalogs().await.expect("ok");
    assert!(client.catalogs_last_error().is_none());
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        Some(8)
    );
}

#[tokio::test]
async fn l2_wait_for_catalogs_can_retry_after_transient_failure() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let attempts_h = attempts.clone();
    let spot = spot_config_fixture();
    let zipper = zipper_config_fixture();
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            if attempts_h.fetch_add(1, Ordering::SeqCst) == 0 {
                hardening_support::HttpScript::Json {
                    status: 503,
                    body: br#"{"error":"temporary"}"#.to_vec(),
                }
            } else {
                connect_proto_ok(&spot)
            }
        } else if req.path == ZIPPER_CONFIG_PATH {
            connect_proto_ok(&zipper)
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_secs(2),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");

    client
        .wait_for_catalogs()
        .await
        .expect_err("first catalog attempt must surface the transient failure");
    client
        .wait_for_catalogs()
        .await
        .expect("second catalog attempt must recover");
    assert!(client.catalogs.is_ready());
    assert!(client.catalogs_last_error().is_none());
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        Some(8)
    );
}

#[tokio::test]
async fn l2_wait_for_catalogs_zipper_failure_leaves_spot_unhydrated() {
    let spot = spot_config_fixture();
    let http = MockHttpServer::spawn(move |req| {
        if req.path == SPOT_CONFIG_PATH {
            connect_proto_ok(&spot)
        } else if req.path == ZIPPER_CONFIG_PATH {
            hardening_support::HttpScript::Json {
                status: 500,
                body: br#"{"error":"zipper down"}"#.to_vec(),
            }
        } else {
            hardening_support::HttpScript::NotFound
        }
    })
    .await;
    let client = Client::new(Config {
        api_url: http.base_url(),
        timeout: Duration::from_secs(2),
        hydrate_catalogs: true,
        ..Default::default()
    })
    .expect("client");
    let err = client
        .wait_for_catalogs()
        .await
        .expect_err("zipper 500 must fail closed");
    assert!(err.to_string().contains("catalog"), "{err}");
    assert_eq!(
        client.catalogs.base_quantity_scale_for_symbol("BTC-USDT"),
        None,
        "spot must not install when zipper fails"
    );
}

#[tokio::test]
async fn l2_snapshot_then_stream_recovery_success_clears_err() {
    use std::sync::Mutex;

    let active = Arc::new(AtomicUsize::new(0));
    let ws = MockWsServer::spawn_centrifugo_public(active.clone()).await;
    let rt = RealtimeClient::new(ws.ws_url(), "", None, None);
    let attempts = Arc::new(AtomicUsize::new(0));
    let fetch_attempts = attempts.clone();
    let merged = Arc::new(Mutex::new(Vec::<u8>::new()));
    let merged_cb = merged.clone();
    let sts = SnapshotThenStream::new(SnapshotThenStreamConfig {
        client: rt,
        channel: PUBLIC_CHANNEL.into(),
        decode: Arc::new(|_b| Ok(1u8)),
        fetch_snapshot: Arc::new(move || {
            let attempt = fetch_attempts.fetch_add(1, Ordering::SeqCst);
            async move {
                if attempt == 0 {
                    Err(Error::transport("snapshot refresh failed"))
                } else {
                    Ok("recovered".to_string())
                }
            }
            .boxed()
        }),
        read_publication: Arc::new(|p| vec![p]),
        apply_snapshot: Arc::new(move |_s, pending| {
            merged_cb.lock().expect("merged").extend(pending);
        }),
        apply_live_publications: Arc::new(|_p| {}),
        max_buffered: 8,
        on_reconnect: None,
        on_snapshot_refresh: None,
        on_error: None,
    });

    // Drive buffer retention without waiting on WS publications: inject via refresh path.
    // First refresh fails (sets err, keeps buffer); inject pubs while not ready; retry succeeds.
    assert!(sts.refresh_snapshot().await.is_err());
    assert!(sts.err().is_some());
    // Direct buffer injection through a second STS that shares apply is awkward; instead
    // re-run the unit-level retention contract via public refresh_snapshot success clear.
    sts.refresh_snapshot().await.expect("recovery");
    assert!(sts.is_ready());
    assert!(sts.err().is_none());
    assert!(attempts.load(Ordering::SeqCst) >= 2);
    sts.close();
}

#[test]
fn withdrawal_payload_signing_and_e18_conversion_are_exact() {
    use buffa::Message;
    use ed25519_dalek::{Signature, Verifier, VerifyingKey};
    use polyester::models::CreateApiKeyTradingWithdrawParams;
    use polyester::{AssetAmount, Client, Config, QuantityDomain};

    let client = Client::new(Config {
        api_key_id: Some("withdraw-hardening".into()),
        api_private_key: Some(TEST_KEY.into()),
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    let prepared = client
        .withdraw
        .prepare_api_key_to_funding(CreateApiKeyTradingWithdrawParams {
            asset_id: 7,
            amount: AssetAmount::from_scaled(125, Some(2), QuantityDomain::LedgerE18, Some(7))
                .unwrap(),
            destination_address: String::new(),
            idempotency_key: "withdraw-hardening".into(),
            amount_scale: Some(2),
            deadline_ts_sec: Some(1_800_000_000),
            nonce: Some(42),
        })
        .unwrap();

    assert_eq!(prepared.payload().deadline_ts_sec, 1_800_000_000);
    assert_eq!(
        prepared.deterministic_payload_bytes(),
        prepared.payload().encode_to_vec()
    );
    let amount = prepared.payload().amount_e18.as_option().unwrap();
    assert_eq!(
        (u128::from(amount.hi) << 64) | u128::from(amount.lo),
        1_250_000_000_000_000_000
    );

    let seed: [u8; 32] = hex::decode(TEST_KEY).unwrap().try_into().unwrap();
    let verifier = VerifyingKey::from(&ed25519_dalek::SigningKey::from_bytes(&seed));
    verifier
        .verify(
            &prepared.deterministic_payload_bytes(),
            &Signature::from_slice(prepared.payload_signature()).unwrap(),
        )
        .unwrap();
}

#[tokio::test]
async fn precomputed_withdraw_fields_and_unknown_wallet_actions_fail_closed() {
    use polyester::models::{CreateTradingWithdrawParams, CreateWalletTradingWithdrawParams};
    use polyester::{AssetAmount, Client, Config, Error, QuantityDomain};

    let client = Client::new(Config {
        hydrate_catalogs: false,
        ..Default::default()
    })
    .unwrap();
    let amount =
        || AssetAmount::from_scaled(1, Some(18), QuantityDomain::LedgerE18, Some(7)).unwrap();
    let missing_deadline = client
        .withdraw
        .create_to_funding(CreateTradingWithdrawParams {
            asset_id: 7,
            amount: amount(),
            payload_signature: vec![1],
            destination_address: String::new(),
            idempotency_key: "missing-deadline".into(),
            amount_scale: Some(18),
            deadline_ts_sec: None,
            nonce: 42,
        })
        .await
        .unwrap_err();
    assert!(matches!(missing_deadline, Error::Validation(_)));

    let unknown_action = client
        .withdraw
        .create_wallet_trading_withdraw(CreateWalletTradingWithdrawParams {
            action: "unknown".into(),
            asset_id: 7,
            amount: amount(),
            idempotency_key: "unknown-action".into(),
            payload_signature: vec![1],
            signer_wallet: "0x1".into(),
            destination_chain_id: 0,
            destination_address: String::new(),
            subaccount_id: None,
            amount_scale: Some(18),
            deadline_ts_sec: Some(1_800_000_000),
            nonce: 42,
        })
        .await
        .unwrap_err();
    assert!(matches!(unknown_action, Error::Validation(_)));
}

// Silence unused import warnings when Credentials helpers evolve.
#[allow(dead_code)]
fn _creds_type_check() -> Credentials {
    test_credentials("ak", TEST_KEY)
}

#[allow(dead_code)]
fn _reply_helper() -> Vec<u8> {
    centrifugo_ok_reply(1)
}