//! End-to-end streaming transport tests.
//!
//! These tests exercise the complete streaming path: operation send-side ->
//! transport fragmentation -> UDP delivery -> transport reassembly -> orphan claim ->
//! operation receive-side -> correct result.
//!
//! Assertions verify that contract state actually arrived in non-gateway node
//! storages with the correct bytes, replacing the earlier global-counter approach.
//!
//! All tests use `run_controlled_simulation()` for deterministic execution via Turmoil.
//!
//! Enable with: cargo test -p freenet --features "simulation_tests,testing" --test streaming_e2e

#![cfg(feature = "simulation_tests")]

use freenet::config::{GlobalRng, GlobalSimulationTime};
use freenet::dev_tool::{
    MockStateStorage, NodeLabel, ScheduledOperation, SimNetwork, SimOperation,
};
use freenet_stdlib::prelude::*;
use std::time::Duration;

// =============================================================================
// Helpers
// =============================================================================

/// Set up a SimNetwork with streaming enabled at the given threshold.
async fn setup_streaming_network(
    name: &str,
    gateways: usize,
    nodes: usize,
    seed: u64,
    streaming_threshold: usize,
) -> SimNetwork {
    GlobalRng::set_seed(seed);
    const BASE_EPOCH_MS: u64 = 1577836800000;
    const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
    GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + (seed % RANGE_MS));

    let mut sim = SimNetwork::new(
        name, gateways, nodes, 7,  // ring_max_htl
        3,  // rnd_if_htl_above
        10, // max_connections
        2,  // min_connections
        seed,
    )
    .await;
    sim.with_streaming_threshold(streaming_threshold);
    sim
}

/// Check that at least one non-gateway node stored the given contract key,
/// and return the stored state bytes if found.
fn find_contract_in_non_gateway_storages(
    node_storages: &std::collections::HashMap<NodeLabel, MockStateStorage>,
    contract_key: &ContractKey,
) -> Option<WrappedState> {
    for (label, storage) in node_storages {
        if label.is_node() {
            if let Some(state) = storage.get_stored_state(contract_key) {
                return Some(state);
            }
        }
    }
    None
}

// =============================================================================
// Test 1: Streaming PUT with large state
// =============================================================================

/// Tests that a large state PUT delivers the correct state to non-gateway nodes.
///
/// - SimNetwork: 1 gateway + 2 nodes, streaming threshold = 1024 bytes
/// - Gateway PUTs a contract with 100KB state (well above threshold)
/// - Asserts: simulation completes, state arrived at a non-gateway node with correct bytes
#[test]
fn test_streaming_put_large_state() {
    const SEED: u64 = 0x5720_0001_DEAD_BEEF;
    const NETWORK_NAME: &str = "streaming-put-large";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 100 * 1024; // 100KB

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 2, SEED, THRESHOLD));

    let contract = SimOperation::create_test_contract(42);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 42);
    let contract_key = contract.key();

    let operations = vec![
        // Gateway PUTs large contract
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: large_state.clone(),
                subscribe: false,
            },
        ),
    ];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(120),
        Duration::from_secs(60),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Streaming PUT simulation should complete: {:?}",
        result.turmoil_result.err()
    );

    let stored = find_contract_in_non_gateway_storages(&result.node_storages, &contract_key);
    assert!(
        stored.is_some(),
        "Expected 100KB contract to be stored in at least one non-gateway node"
    );
    let stored_bytes: Vec<u8> = stored.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, large_state,
        "Stored state bytes should match the original 100KB state"
    );
}

// =============================================================================
// Test 2: Below threshold uses inline
// =============================================================================

/// Tests that a small state below the streaming threshold still delivers correctly.
///
/// - Same setup but state = 512 bytes (below 1024 threshold)
/// - Asserts: state arrived at a non-gateway node with correct bytes
#[test]
fn test_streaming_put_below_threshold_uses_inline() {
    const SEED: u64 = 0x1011_0002_CAFE_BABE;
    const NETWORK_NAME: &str = "streaming-inline";
    const THRESHOLD: usize = 1024;
    const SMALL_STATE_SIZE: usize = 512; // Below threshold

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 2, SEED, THRESHOLD));

    let contract = SimOperation::create_test_contract(99);
    let small_state = SimOperation::create_large_state(SMALL_STATE_SIZE, 99);
    let contract_key = contract.key();

    let operations = vec![ScheduledOperation::new(
        NodeLabel::gateway(NETWORK_NAME, 0),
        SimOperation::Put {
            contract: contract.clone(),
            state: small_state.clone(),
            subscribe: false,
        },
    )];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(120),
        Duration::from_secs(45),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Inline PUT simulation should complete: {:?}",
        result.turmoil_result.err()
    );

    let stored = find_contract_in_non_gateway_storages(&result.node_storages, &contract_key);
    assert!(
        stored.is_some(),
        "Expected 512-byte contract to be stored in at least one non-gateway node"
    );
    let stored_bytes: Vec<u8> = stored.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, small_state,
        "Stored state bytes should match the original 512-byte state"
    );
}

// =============================================================================
// Test 3: Streaming UPDATE broadcast
// =============================================================================

/// Tests that streaming UPDATE broadcasts deliver the updated state.
///
/// - SimNetwork: 1 gateway + 3 nodes, streaming threshold = 1024
/// - Gateway PUTs contract with small state + subscribe
/// - Nodes subscribe
/// - Gateway UPDATEs with large state
/// - Asserts: at least one subscribing node has the updated state
#[test]
fn test_streaming_update_broadcast() {
    const SEED: u64 = 0xBCA5_0003_1234_5678;
    const NETWORK_NAME: &str = "streaming-update";
    const THRESHOLD: usize = 1024;
    const LARGE_UPDATE_SIZE: usize = 100 * 1024; // 100KB

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 3, SEED, THRESHOLD));

    let contract = SimOperation::create_test_contract(55);
    let contract_key = contract.key();
    let initial_state = SimOperation::create_test_state(55);
    let large_update = SimOperation::create_large_state(LARGE_UPDATE_SIZE, 77);

    let operations = vec![
        // Gateway PUTs contract with subscribe
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: initial_state,
                subscribe: true,
            },
        ),
        // Node 1 subscribes
        ScheduledOperation::new(
            NodeLabel::node(NETWORK_NAME, 1),
            SimOperation::Subscribe {
                contract_id: *contract_key.id(),
            },
        ),
        // Node 2 subscribes
        ScheduledOperation::new(
            NodeLabel::node(NETWORK_NAME, 2),
            SimOperation::Subscribe {
                contract_id: *contract_key.id(),
            },
        ),
        // Gateway UPDATEs with large state
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Update {
                key: contract_key,
                data: large_update,
            },
        ),
    ];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(120),
        Duration::from_secs(60),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Streaming UPDATE broadcast simulation should complete: {:?}",
        result.turmoil_result.err()
    );

    // At least one non-gateway node should have a state for this contract
    // (the update may have been applied as delta or full state depending on runtime)
    let stored = find_contract_in_non_gateway_storages(&result.node_storages, &contract_key);
    assert!(
        stored.is_some(),
        "Expected at least one subscribing non-gateway node to have state for the updated contract"
    );
}

// =============================================================================
// Test 4: Multiple concurrent streaming PUTs
// =============================================================================

/// Tests that multiple concurrent large PUTs deliver the correct state for each contract.
///
/// - SimNetwork: 1 gateway + 2 nodes, streaming threshold = 1024
/// - Gateway PUTs contract A (50KB) and contract B (80KB)
/// - Asserts: both contracts found in non-gateway storage with correct state bytes
#[test]
fn test_streaming_multiple_concurrent_puts() {
    const SEED: u64 = 0xC00C_0004_ABCD_EF01;
    const NETWORK_NAME: &str = "streaming-concurrent";
    const THRESHOLD: usize = 1024;

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 2, SEED, THRESHOLD));

    let contract_a = SimOperation::create_test_contract(10);
    let state_a = SimOperation::create_large_state(50 * 1024, 10); // 50KB
    let key_a = contract_a.key();

    let contract_b = SimOperation::create_test_contract(20);
    let state_b = SimOperation::create_large_state(80 * 1024, 20); // 80KB
    let key_b = contract_b.key();

    let operations = vec![
        // Gateway PUTs contract A
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract_a.clone(),
                state: state_a.clone(),
                subscribe: false,
            },
        ),
        // Gateway PUTs contract B
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract_b.clone(),
                state: state_b.clone(),
                subscribe: false,
            },
        ),
    ];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(120),
        Duration::from_secs(60),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Multiple concurrent streaming PUTs should complete: {:?}",
        result.turmoil_result.err()
    );

    // Verify contract A
    let stored_a = find_contract_in_non_gateway_storages(&result.node_storages, &key_a);
    assert!(
        stored_a.is_some(),
        "Expected 50KB contract A to be stored in at least one non-gateway node"
    );
    let stored_a_bytes: Vec<u8> = stored_a.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_a_bytes, state_a,
        "Stored state bytes for contract A should match the original 50KB state"
    );

    // Verify contract B
    let stored_b = find_contract_in_non_gateway_storages(&result.node_storages, &key_b);
    assert!(
        stored_b.is_some(),
        "Expected 80KB contract B to be stored in at least one non-gateway node"
    );
    let stored_b_bytes: Vec<u8> = stored_b.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_b_bytes, state_b,
        "Stored state bytes for contract B should match the original 80KB state"
    );
}

// =============================================================================
// Test 5: Streaming with packet loss
// =============================================================================

/// Tests that streaming transport handles packet loss gracefully.
///
/// - SimNetwork with 5% message loss rate
/// - Large state PUT (100KB, well above 1024 threshold)
/// - Asserts: contract arrived at non-gateway node with correct state bytes
#[test]
fn test_streaming_with_packet_loss() {
    use freenet::simulation::FaultConfig;

    const SEED: u64 = 0xA055_0005_BEEF_CAFE;
    const NETWORK_NAME: &str = "streaming-lossy";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 100 * 1024; // 100KB

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let mut sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 2, SEED, THRESHOLD));

    // Inject 5% message loss
    let fault_config = FaultConfig::builder().message_loss_rate(0.05).build();
    sim.with_fault_injection(fault_config);

    let contract = SimOperation::create_test_contract(33);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 33);
    let contract_key = contract.key();

    let operations = vec![
        // Gateway PUTs large contract
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: large_state.clone(),
                subscribe: false,
            },
        ),
    ];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(180), // Longer timeout for lossy network
        Duration::from_secs(90),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Streaming with 5% packet loss should still complete: {:?}",
        result.turmoil_result.err()
    );

    let stored = find_contract_in_non_gateway_storages(&result.node_storages, &contract_key);
    assert!(
        stored.is_some(),
        "Expected 100KB contract to be stored in at least one non-gateway node even with packet loss"
    );
    let stored_bytes: Vec<u8> = stored.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, large_state,
        "Stored state bytes should match the original 100KB state despite packet loss"
    );
}

// =============================================================================
// Test 6: Streaming with packet reordering
// =============================================================================

/// Tests that streaming handles packet reordering via variable latency.
///
/// - SimNetwork: 1 gateway + 2 nodes, streaming threshold = 1024
/// - Fault injection: variable latency 10ms..100ms (causes reordering)
/// - Gateway PUTs a 100KB contract
/// - Asserts: correct state arrives despite fragment reordering
///
/// LockFreeStreamBuffer indexes by fragment number, so reordering should be
/// handled transparently.
#[test]
fn test_streaming_with_packet_reordering() {
    use freenet::simulation::FaultConfig;

    const SEED: u64 = 0xBE0F_0006_DEAD_1234;
    const NETWORK_NAME: &str = "streaming-reorder";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 100 * 1024; // 100KB

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let mut sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 2, SEED, THRESHOLD));

    // Inject variable latency to cause packet reordering
    let fault_config = FaultConfig::builder()
        .latency_range(Duration::from_millis(10)..Duration::from_millis(100))
        .build();
    sim.with_fault_injection(fault_config);

    let contract = SimOperation::create_test_contract(66);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 66);
    let contract_key = contract.key();

    let operations = vec![ScheduledOperation::new(
        NodeLabel::gateway(NETWORK_NAME, 0),
        SimOperation::Put {
            contract: contract.clone(),
            state: large_state.clone(),
            subscribe: false,
        },
    )];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(180),
        Duration::from_secs(90),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Streaming with packet reordering should complete: {:?}",
        result.turmoil_result.err()
    );

    let stored = find_contract_in_non_gateway_storages(&result.node_storages, &contract_key);
    assert!(
        stored.is_some(),
        "Expected 100KB contract to be stored despite packet reordering"
    );
    let stored_bytes: Vec<u8> = stored.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, large_state,
        "Stored state bytes should match the original 100KB state despite reordering"
    );
}

// =============================================================================
// Test 7: Multi-hop forwarding
// =============================================================================

/// Tests streaming through multi-hop forwarding with a larger network.
///
/// - SimNetwork: 1 gateway + 4 nodes, streaming threshold = 1024
/// - Gateway PUTs a 200KB contract
/// - With 4 nodes, the probability of multi-hop routing is higher
/// - Asserts: at least one non-gateway node received the contract
#[test]
fn test_streaming_multi_hop_forwarding() {
    const SEED: u64 = 0xF1A7_0007_CAFE_9876;
    const NETWORK_NAME: &str = "streaming-multihop";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 200 * 1024; // 200KB

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 4, SEED, THRESHOLD));

    let contract = SimOperation::create_test_contract(88);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 88);
    let contract_key = contract.key();

    let operations = vec![ScheduledOperation::new(
        NodeLabel::gateway(NETWORK_NAME, 0),
        SimOperation::Put {
            contract: contract.clone(),
            state: large_state.clone(),
            subscribe: false,
        },
    )];

    // #1454 phase 5 slice B: snapshot the streaming relay driver
    // call count so we can assert the task-per-tx path handled at
    // least one relay hop (vs. legacy `handle_op_request` fallthrough).
    let driver_calls_before = freenet::dev_tool::RELAY_PUT_STREAMING_DRIVER_CALL_COUNT
        .load(std::sync::atomic::Ordering::SeqCst);

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(180),
        Duration::from_secs(90),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Multi-hop streaming PUT should complete: {:?}",
        result.turmoil_result.err()
    );

    let driver_calls_after = freenet::dev_tool::RELAY_PUT_STREAMING_DRIVER_CALL_COUNT
        .load(std::sync::atomic::Ordering::SeqCst);
    assert!(
        driver_calls_after > driver_calls_before,
        "Expected the streaming PUT relay driver to run at least once \
         during a 200KB multi-hop PUT (before={driver_calls_before}, \
         after={driver_calls_after}). Counter stuck → slice B dispatch \
         gate fell through to legacy `handle_op_request`."
    );

    let stored = find_contract_in_non_gateway_storages(&result.node_storages, &contract_key);
    assert!(
        stored.is_some(),
        "Expected 200KB contract to be stored in at least one non-gateway node via multi-hop"
    );
    let stored_bytes: Vec<u8> = stored.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, large_state,
        "Stored state bytes should match the original 200KB state"
    );
}

// =============================================================================
// Test 8: Streaming GET through relay hop
// =============================================================================

/// Tests that a large contract can be retrieved via streaming GET through relay nodes.
///
/// This exercises the relay pipe_stream path, which was the code path responsible
/// for the streaming hang bug (#3608). The scenario:
///
/// 1. Gateway PUTs a ~1MB contract (stored at nodes near its ring location)
/// 2. A different node GETs the contract, routing through intermediate relay nodes
/// 3. The relay uses pipe_stream to forward the streaming GET response
///
/// With a sparse topology (1 gateway + 6 nodes, max_connections = 3, below
/// the node count) the GET request from node 5 cannot mesh directly with the
/// storer and must relay through at least one intermediate node, exercising
/// the pipe_stream forwarding path that the cwnd timeout protects. The
/// `RELAY_GET_STREAMING_FORWARD_COUNT` assertion below proves a streaming
/// relay hop genuinely fired (the sim is deterministic per-seed, so this is
/// not flaky once the seed + topology are locked in).
#[test]
fn test_streaming_get_through_relay() {
    const SEED: u64 = 0xDE1A_0008_FACE_B00C;
    const NETWORK_NAME: &str = "streaming-get-relay";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 1024 * 1024; // ~1MB, similar to River UI container

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    // Sparse topology: 1 gateway + 6 nodes, max_connections = 3 (below the
    // node count) forces the getter (node 5) to relay rather than connect
    // directly to the storer. `setup_streaming_network` hardcodes
    // max_connections = 10, which meshes a network this small and yields a
    // direct GET with no relay hop (counter stays 0), so build the
    // SimNetwork inline here with the sparse cap.
    GlobalRng::set_seed(SEED);
    const BASE_EPOCH_MS: u64 = 1577836800000;
    const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
    GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + (SEED % RANGE_MS));
    let mut sim = rt.block_on(async {
        let mut sim = SimNetwork::new(
            NETWORK_NAME,
            1, // gateways
            6, // nodes
            7, // ring_max_htl
            3, // rnd_if_htl_above
            3, // max_connections (sparse: < node count → forces a relay hop)
            2, // min_connections
            SEED,
        )
        .await;
        sim.with_streaming_threshold(THRESHOLD);
        sim
    });
    // This test asserts a streaming relay-forward hop fires (node 5 must relay
    // to reach the storer). Placement migration (#4404) spreads the contract
    // toward the requester, so the relay hop never runs and the
    // RELAY_GET_STREAMING_FORWARD_COUNT assertion fails. The test exercises the
    // relay-streaming mechanic, not placement, so pin migration OFF — don't
    // rely on build-version gating, which flips ON on the v0.2.73 release
    // branch where #4404 ships active.
    sim.disable_placement_migration();

    let contract = SimOperation::create_test_contract(0xAB);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 0xAB);
    let contract_key = contract.key();
    let contract_id = *contract_key.id();

    let operations = vec![
        // Gateway PUTs 1MB contract
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: large_state.clone(),
                subscribe: false,
            },
        ),
        // Node 5 GETs the contract — must relay through network
        ScheduledOperation::new(
            NodeLabel::node(NETWORK_NAME, 5),
            SimOperation::Get {
                contract_id,
                return_contract_code: false,
                subscribe: false,
            },
        ),
    ];

    // Snapshot the streaming relay GET forward counter so we can assert the
    // relay genuinely forked+piped a streaming response through a hop (vs.
    // satisfying the GET via store-and-forward without the streaming path).
    let forward_calls_before = freenet::dev_tool::RELAY_GET_STREAMING_FORWARD_COUNT
        .load(std::sync::atomic::Ordering::SeqCst);

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(300), // Longer timeout for 1MB streaming GET
        Duration::from_secs(120),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Streaming GET through relay should complete: {:?}",
        result.turmoil_result.err()
    );

    let forward_calls_after = freenet::dev_tool::RELAY_GET_STREAMING_FORWARD_COUNT
        .load(std::sync::atomic::Ordering::SeqCst);
    assert!(
        forward_calls_after > forward_calls_before,
        "Expected the relay GET streaming-forward path to run at least once \
         during a 1MB relayed GET (before={forward_calls_before}, \
         after={forward_calls_after}). Counter stuck → relay GET \
         streaming-forward path did not run (fell back to store-and-forward)."
    );

    // The GET-requesting node should have the contract state
    let getter_label = NodeLabel::node(NETWORK_NAME, 5);
    let getter_storage = result
        .node_storages
        .get(&getter_label)
        .expect("node 5 should have a storage handle");
    let getter_state = getter_storage.get_stored_state(&contract_key);

    assert!(
        getter_state.is_some(),
        "Node 5 should have 1MB contract state after streaming GET through relay"
    );
    let stored_bytes: Vec<u8> = getter_state.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, large_state,
        "Stored state bytes should match the original 1MB state after relay GET"
    );
}

/// Regression test: 1MB streaming GET with packet loss.
///
/// This is the test that was missing and would have caught the loss_pause margin
/// bug in v0.2.22. The existing tests covered:
/// - 100KB streaming with 5% loss (too small for stall to manifest)
/// - 1MB streaming GET without loss (no loss_pause triggered)
///
/// This test combines both: a 1MB transfer that triggers streaming, with 5%
/// packet loss that triggers loss_pause recovery. With the old 2-packet margin,
/// this test would timeout because the sender stalls for 20s per loss event.
/// With the 50-packet margin, recovery completes quickly.
#[test]
fn test_streaming_get_1mb_with_packet_loss() {
    use freenet::simulation::FaultConfig;

    const SEED: u64 = 0xDE1A_000A_DEAD_BEEF;
    const NETWORK_NAME: &str = "streaming-get-lossy-1mb";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 1024 * 1024; // 1MB — similar to River UI contract

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let mut sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 4, SEED, THRESHOLD));

    // 5% message loss — enough to trigger loss_pause during the ~833-packet transfer
    let fault_config = FaultConfig::builder().message_loss_rate(0.05).build();
    sim.with_fault_injection(fault_config);

    let contract = SimOperation::create_test_contract(0xDF);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 0xDF);
    let contract_key = contract.key();
    let contract_id = *contract_key.id();

    let operations = vec![
        // Gateway PUTs 1MB contract
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: large_state.clone(),
                subscribe: false,
            },
        ),
        // A different node GETs the contract — must relay through network
        ScheduledOperation::new(
            NodeLabel::node(NETWORK_NAME, 3),
            SimOperation::Get {
                contract_id,
                return_contract_code: false,
                subscribe: false,
            },
        ),
    ];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(300), // Generous timeout for lossy 1MB transfer
        Duration::from_secs(120),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "1MB streaming GET with 5% packet loss should complete without stalling: {:?}",
        result.turmoil_result.err()
    );

    // The GET-requesting node should have the correct 1MB state
    let node3_label = NodeLabel::node(NETWORK_NAME, 3);
    let node3_storage = result
        .node_storages
        .get(&node3_label)
        .expect("node 3 should have a storage handle");
    let node3_state = node3_storage.get_stored_state(&contract_key);

    assert!(
        node3_state.is_some(),
        "Node 3 should have 1MB contract state after streaming GET with packet loss"
    );
    let stored_bytes: Vec<u8> = node3_state.unwrap().as_ref().to_vec();
    assert_eq!(
        stored_bytes, large_state,
        "Stored state bytes should match the original 1MB state despite packet loss"
    );
}

/// Regression test for #3704: streaming GET path must HOST the fetched
/// contract on the return path (call `announce_contract_hosted()`, not just
/// `record_get_access()`) so subsequent GETs can be served from local cache.
///
/// NOTE: piece E of the demand-driven hosting redesign REMOVED GET-auto-subscribe.
/// The GETting node hosts under piece A's demand gauge (evictable, non-durable),
/// but it does NOT auto-install a durable subscription. This test therefore
/// asserts only the hosting side effect (`is_hosting`), not subscription — a GET
/// with subscribe=false must not subscribe (see hosting-invariants, invariants
/// 1 & 2, and `test_driver_inline_get_does_not_auto_subscribe`).
#[test]
fn test_streaming_get_hosts_on_return_path() {
    const SEED: u64 = 0xDE1A_0009_CAFE_F00D;
    const NETWORK_NAME: &str = "streaming-get-hosts-return-path";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 100 * 1024; // 100KB, above streaming threshold

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let sim = rt.block_on(setup_streaming_network(NETWORK_NAME, 1, 3, SEED, THRESHOLD));

    let contract = SimOperation::create_test_contract(0xCC);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 0xCC);
    let contract_key = contract.key();
    let contract_id = *contract_key.id();

    let operations = vec![
        // Gateway PUTs 100KB contract
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: large_state.clone(),
                subscribe: true,
            },
        ),
        // Node 2 GETs the contract (will be streamed since >1KB threshold).
        // subscribe=false: the node hosts the fetched state on the return path
        // (under the demand gauge) but must NOT auto-subscribe (piece E).
        ScheduledOperation::new(
            NodeLabel::node(NETWORK_NAME, 2),
            SimOperation::Get {
                contract_id,
                return_contract_code: true,
                subscribe: false,
            },
        ),
    ];

    let result = sim.run_controlled_simulation(
        SEED,
        operations,
        Duration::from_secs(300),
        Duration::from_secs(120),
    );

    assert!(
        result.turmoil_result.is_ok(),
        "Streaming GET should complete: {:?}",
        result.turmoil_result.err()
    );

    // Verify the GET-requesting node has the contract state
    let node2_label = NodeLabel::node(NETWORK_NAME, 2);
    let node2_storage = result
        .node_storages
        .get(&node2_label)
        .expect("node 2 should have a storage handle");
    let node2_state = node2_storage.get_stored_state(&contract_key);
    assert!(
        node2_state.is_some(),
        "Node 2 should have contract state after streaming GET"
    );

    // Key assertion: the GET-requesting node should be HOSTING the contract
    // (i.e., announce_contract_hosted was called, not just record_get_access)
    let node2_snapshot = result.topology_snapshots.iter().find(|s| {
        s.contracts
                .values()
                .any(|c| c.contract_key == contract_key && c.is_hosting)
                // Match by checking this is node 2's snapshot (not gateway)
                && s.peer_addr.ip() != std::net::IpAddr::from([1u8, 0, 0, 1])
    });

    assert!(
        node2_snapshot.is_some(),
        "After streaming GET, the requesting node should be hosting the contract. \
         Topology snapshots: {:#?}",
        result
            .topology_snapshots
            .iter()
            .map(|s| (
                s.peer_addr,
                s.contracts
                    .values()
                    .map(|c| (&c.contract_key, c.is_hosting, c.upstream.is_some()))
                    .collect::<Vec<_>>()
            ))
            .collect::<Vec<_>>()
    );

    // NOTE on subscription: this test asserts ONLY host-on-GET (`is_hosting`),
    // not the absence of a subscription. In this topology the gateway's
    // `subscribe=true` PUT propagates the contract to its neighbors and builds a
    // subscription tree, so the GETting node ends up subscribed via that path
    // regardless of GET-auto-subscribe. The piece-E guarantee — that the GET
    // *driver* itself no longer auto-subscribes on a `subscribe=false` GET — is
    // pinned at the source level by `driver_does_not_auto_subscribe_on_get_response`
    // (runs) and behaviorally isolated (HTL=1, no PUT propagation) by
    // `test_driver_inline_get_does_not_auto_subscribe` (currently `#[ignore]`'d
    // for the pre-existing #3883 topology-routing gap).
}

// =============================================================================
// Regression test for #1454 Phase 3b — driver must handle streaming GETs.
//
// The existing streaming tests above all satisfy the GETting node's request
// via the client-events local-cache shortcut (`client_events.rs:1108-1154`)
// because the gateway's `SimOperation::Put` propagates the contract to
// neighbors during PUT, leaving the GETting node with a relay-cached copy
// BEFORE the GET fires. That shortcut returns without ever calling
// `start_client_get`, so the driver's `Terminal::Streaming` path is
// untested.
//
// This test uses `SimOperation::SeedContract`, which seeds only the gateway's
// local store without any network propagation. Node 3 then cold-GETs — the
// local-cache shortcut misses, the request flows through the task-per-tx
// driver, and the terminal reply arrives as `ResponseStreaming` because the
// payload is above the streaming threshold. The driver's `Done` arm must
// produce a client-visible `HostResponse::GetResponse` with the correct
// state AND write the state to node 3's local store for subsequent access.
//
// Without the fix, `Terminal::Streaming` in the driver does not write the
// store (the bypass skips `process_message`, and
// `stream_handle.assemble()` never runs on the originator under
// task-per-tx), so `build_host_response`'s re-query returns `None` and the
// client receives an `OperationError`. Node 3's storage check then fails.
// =============================================================================

/// Set up a SimNetwork with a LOW ring_max_htl so PUT fan-out can't
/// reach the GETting node — the setup that forces a cold-cache GET
/// through the task-per-tx driver.
///
/// `setup_streaming_network` uses `ring_max_htl=7` which is enough
/// for PUT to propagate to every node in small topologies, making the
/// GETter's `client_events.rs` local-cache shortcut always satisfy
/// the request.
async fn setup_htl1_network(name: &str, nodes: usize, seed: u64, threshold: usize) -> SimNetwork {
    GlobalRng::set_seed(seed);
    const BASE_EPOCH_MS: u64 = 1577836800000;
    const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
    GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + (seed % RANGE_MS));

    let mut sim = SimNetwork::new(
        name, 1,     // 1 gateway
        nodes, // N nodes
        1,     // ring_max_htl = 1 — PUT forwards at most once past the originator
        1,     // rnd_if_htl_above
        10,    // max_connections
        2,     // min_connections
        seed,
    )
    .await;
    sim.with_streaming_threshold(threshold);
    sim
}

/// Cold-cache streaming GET via task-per-tx driver.
///
/// Scope: bug #1 from the #3884 skeptical review — the driver's
/// `Terminal::Streaming` path does not write the contract state to the
/// local store, so any client GET of a >threshold contract from a node
/// that has no relay-cached copy returns `OperationError` on the client
/// channel and leaves local storage empty.
///
/// ## Driver isolation strategy
///
/// Uses `ring_max_htl = 1` so the gateway's PUT propagates at most one
/// hop. In a 10-node topology, at least one node (the one furthest
/// from the gateway) doesn't receive the state during PUT. That node's
/// cold GET falls through the `client_events.rs:1108` shortcut
/// (since `local_satisfies_request = false`) and invokes the
/// task-per-tx driver. Verification via `GET_DRIVER_CALL_COUNT`.
///
/// ## Known infrastructure gap
///
/// Currently `#[ignore]`'d because even with the driver correctly
/// reached, the HTL=1 topology used to force cold-cache leaves the
/// driver unable to route (driver retries exhaust with "channel
/// closed"). The driver fix itself is validated by unit tests in
/// `operations/get/op_ctx_task.rs`; the remaining work is test
/// infrastructure (either a SimNetwork helper that sets up a more
/// routable cold-cache topology, or a `SimOperation::SeedContract`
/// variant that also announces hosting). Tracked in #3883.
#[ignore = "Infrastructure gap — HTL=1 topology isn't routable; see docstring and #3883"]
#[test]
fn test_driver_streaming_get_cold_cache() {
    use freenet::dev_tool::GET_DRIVER_CALL_COUNT;
    use std::sync::atomic::Ordering;

    const SEED: u64 = 0xDE1A_0B0B_C01D_CA7E;
    const NETWORK_NAME_PHASE1: &str = "driver-streaming-cold-cache-p1";
    const NETWORK_NAME_PHASE2: &str = "driver-streaming-cold-cache-p2";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 100 * 1024; // 100KB, above THRESHOLD
    const NODES: usize = 10;

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let contract = SimOperation::create_test_contract(0xBC);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 0xBC);
    let contract_key = contract.key();
    let contract_id = *contract_key.id();

    // Phase 1: run the PUT in isolation to discover which nodes
    // DIDN'T receive the state during PUT fan-out (with HTL=1 and
    // 10 nodes, at least one should be cold).
    let sim1 = rt.block_on(setup_htl1_network(
        NETWORK_NAME_PHASE1,
        NODES,
        SEED,
        THRESHOLD,
    ));
    let phase1_ops = vec![ScheduledOperation::new(
        NodeLabel::gateway(NETWORK_NAME_PHASE1, 0),
        SimOperation::Put {
            contract: contract.clone(),
            state: large_state.clone(),
            subscribe: false,
        },
    )];
    let phase1 = sim1.run_controlled_simulation(
        SEED,
        phase1_ops,
        Duration::from_secs(120),
        Duration::from_secs(30),
    );
    assert!(
        phase1.turmoil_result.is_ok(),
        "Phase 1 (PUT only) should complete: {:?}",
        phase1.turmoil_result.err()
    );
    let cold_node_idx = (1..=NODES).find(|i| {
        let label = NodeLabel::node(NETWORK_NAME_PHASE1, *i);
        phase1
            .node_storages
            .get(&label)
            .is_some_and(|s| s.get_stored_state(&contract_key).is_none())
    });
    let Some(cold_idx) = cold_node_idx else {
        panic!(
            "Test precondition: no cold-cache node exists after HTL=1 PUT \
             in {NODES}-node topology. All nodes received the state during \
             fan-out, which contradicts HTL=1. Try a larger topology."
        );
    };

    // Phase 2: fresh network with the same topology (same seed →
    // same ring), gateway PUTs then the cold node GETs. The cold
    // node's local cache is empty → `local_satisfies_request=false`
    // → `client_events.rs` falls through to the driver.
    let baseline_calls = GET_DRIVER_CALL_COUNT.load(Ordering::SeqCst);
    let sim2 = rt.block_on(setup_htl1_network(
        NETWORK_NAME_PHASE2,
        NODES,
        SEED,
        THRESHOLD,
    ));
    let cold_node_label = NodeLabel::node(NETWORK_NAME_PHASE2, cold_idx);
    let phase2_ops = vec![
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME_PHASE2, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: large_state.clone(),
                subscribe: false,
            },
        ),
        ScheduledOperation::new(
            cold_node_label.clone(),
            SimOperation::Get {
                contract_id,
                return_contract_code: true,
                subscribe: false,
            },
        ),
    ];
    let phase2 = sim2.run_controlled_simulation(
        SEED,
        phase2_ops,
        Duration::from_secs(300),
        Duration::from_secs(120),
    );
    assert!(
        phase2.turmoil_result.is_ok(),
        "Phase 2 (PUT + cold GET) should complete: {:?}",
        phase2.turmoil_result.err()
    );

    // Driver-isolation probe: confirm the GET actually invoked the
    // task-per-tx driver (and wasn't short-circuited by
    // `client_events.rs`'s local-cache path).
    let driver_calls = GET_DRIVER_CALL_COUNT.load(Ordering::SeqCst);
    assert!(
        driver_calls > baseline_calls,
        "Test infrastructure failure: GET_DRIVER_CALL_COUNT did not advance during \
         phase 2. Either node {cold_idx} still had a relay-cached copy of the \
         contract (phase-1 topology diverged from phase-2 topology), or the \
         local-cache shortcut fired for some other reason. Without driver \
         invocation this test does not exercise bug #1 from PR #3884."
    );

    // Bug #1 regression: after the driver-routed streaming GET, the
    // cold-cache node must have stored the contract state locally.
    let storage = phase2
        .node_storages
        .get(&cold_node_label)
        .expect("cold-cache node should have a storage handle");
    let stored = storage.get_stored_state(&contract_key);
    assert!(
        stored.is_some(),
        "Bug #1 regression: cold-cache streaming GET through the driver did \
         not write the contract state to local storage on {cold_node_label:?}. \
         The driver's Terminal::Streaming path must ensure local caching."
    );
    assert_eq!(
        stored.unwrap().as_ref().to_vec(),
        large_state,
        "Stored state on {cold_node_label:?} must match the PUT state"
    );
}

/// Cold-cache non-streaming GET must NOT auto-subscribe at the originator
/// (piece E, demand-driven hosting).
///
/// Scope: piece E removed GET-auto-subscribe. A client GET with
/// `subscribe=false` against a cold cache HOSTS the fetched state on the
/// return path (under piece A's demand gauge — the node stores the state) but
/// must NOT install a durable subscription. Auto-subscribing on GET is the
/// anti-pattern (hosting-invariants, invariants 1 & 2). A client that wants
/// ongoing freshness sets `subscribe=true` explicitly.
///
/// ## Driver isolation strategy
///
/// Same HTL=1 approach as `test_driver_streaming_get_cold_cache`.
/// Phase 1 discovers a cold node; phase 2 GETs from it with a small
/// (inline, not streamed) payload; asserts that the driver fired
/// (via `GET_DRIVER_CALL_COUNT`), that the cold node STORED the state
/// (host-on-GET), and that NO node ended up with an active subscription.
///
/// Same infrastructure gap as above — currently `#[ignore]`'d.
#[ignore = "Infrastructure gap — HTL=1 topology isn't routable; see docstring and #3883"]
#[test]
fn test_driver_inline_get_does_not_auto_subscribe() {
    use freenet::dev_tool::GET_DRIVER_CALL_COUNT;
    use std::sync::atomic::Ordering;

    const SEED: u64 = 0xDE1A_0B0C_A570_5C2B;
    const NETWORK_NAME_PHASE1: &str = "driver-inline-auto-sub-p1";
    const NETWORK_NAME_PHASE2: &str = "driver-inline-auto-sub-p2";
    const THRESHOLD: usize = 1024;
    // State well below THRESHOLD — forces inline Response, not streaming.
    const SMALL_STATE_SIZE: usize = 128;
    const NODES: usize = 10;

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let contract = SimOperation::create_test_contract(0xBD);
    let small_state = SimOperation::create_large_state(SMALL_STATE_SIZE, 0xBD);
    let contract_key = contract.key();
    let contract_id = *contract_key.id();

    // Phase 1: discover which node doesn't receive the PUT.
    let sim1 = rt.block_on(setup_htl1_network(
        NETWORK_NAME_PHASE1,
        NODES,
        SEED,
        THRESHOLD,
    ));
    let phase1_ops = vec![ScheduledOperation::new(
        NodeLabel::gateway(NETWORK_NAME_PHASE1, 0),
        SimOperation::Put {
            contract: contract.clone(),
            state: small_state.clone(),
            subscribe: false,
        },
    )];
    let phase1 = sim1.run_controlled_simulation(
        SEED,
        phase1_ops,
        Duration::from_secs(120),
        Duration::from_secs(30),
    );
    assert!(
        phase1.turmoil_result.is_ok(),
        "Phase 1 (PUT only) should complete: {:?}",
        phase1.turmoil_result.err()
    );
    let cold_node_idx = (1..=NODES).find(|i| {
        let label = NodeLabel::node(NETWORK_NAME_PHASE1, *i);
        phase1
            .node_storages
            .get(&label)
            .is_some_and(|s| s.get_stored_state(&contract_key).is_none())
    });
    let Some(cold_idx) = cold_node_idx else {
        panic!(
            "Test precondition: no cold-cache node exists after HTL=1 PUT \
             in {NODES}-node topology."
        );
    };

    // Phase 2: fresh network, gateway PUTs + cold node GETs
    // (subscribe=false, inline payload).
    let baseline_calls = GET_DRIVER_CALL_COUNT.load(Ordering::SeqCst);
    let sim2 = rt.block_on(setup_htl1_network(
        NETWORK_NAME_PHASE2,
        NODES,
        SEED,
        THRESHOLD,
    ));
    let cold_node_label = NodeLabel::node(NETWORK_NAME_PHASE2, cold_idx);
    let phase2_ops = vec![
        ScheduledOperation::new(
            NodeLabel::gateway(NETWORK_NAME_PHASE2, 0),
            SimOperation::Put {
                contract: contract.clone(),
                state: small_state.clone(),
                subscribe: false,
            },
        ),
        ScheduledOperation::new(
            cold_node_label.clone(),
            SimOperation::Get {
                contract_id,
                return_contract_code: true,
                subscribe: false,
            },
        ),
    ];
    let phase2 = sim2.run_controlled_simulation(
        SEED,
        phase2_ops,
        Duration::from_secs(180),
        Duration::from_secs(90),
    );
    assert!(
        phase2.turmoil_result.is_ok(),
        "Phase 2 (PUT + cold GET) should complete: {:?}",
        phase2.turmoil_result.err()
    );

    // Driver-isolation probe.
    let driver_calls = GET_DRIVER_CALL_COUNT.load(Ordering::SeqCst);
    assert!(
        driver_calls > baseline_calls,
        "Test infrastructure failure: GET_DRIVER_CALL_COUNT did not advance \
         during phase 2. The cold node {cold_idx} must have routed through \
         the task-per-tx driver."
    );

    // Precondition: the driver-routed GET stored state locally.
    let storage = phase2
        .node_storages
        .get(&cold_node_label)
        .expect("cold-cache node should have a storage handle");
    assert!(
        storage.get_stored_state(&contract_key).is_some(),
        "Cold-cache node should have stored the contract state after the GET"
    );

    // Piece E (demand-driven hosting): GET-auto-subscribe was REMOVED. A
    // driver-routed GET with subscribe=false must NOT install a durable
    // subscription anywhere — the node still HOSTS the fetched state on the
    // return path (asserted above: it stored the state, under piece A's demand
    // gauge), but no client requested ongoing freshness, so no subscription is
    // created. This inverts the former "Bug #2" assertion: auto-subscribe on
    // GET is the anti-pattern (hosting-invariants, invariants 1 & 2). Neither
    // the PUTting gateway nor the cold getter set subscribe=true, so NO node
    // should carry an active subscription for this contract.
    let any_subscribed = phase2
        .topology_snapshots
        .iter()
        .any(|s| s.active_subscription_keys.contains(&contract_id));

    assert!(
        !any_subscribed,
        "GET-auto-subscribe regression: a GET with subscribe=false installed a \
         durable subscription. Piece E removed AUTO_SUBSCRIBE_ON_GET — a GET \
         hosts under the demand gauge but must NOT auto-subscribe. Active \
         subscriptions: {:#?}",
        phase2
            .topology_snapshots
            .iter()
            .map(|s| (s.peer_addr, s.active_subscription_keys.clone()))
            .collect::<Vec<_>>()
    );
}

// =============================================================================
// Test: GET retries after stream-assembly failure (#4345)
// =============================================================================

/// Build the sparse-but-routable topology used by
/// `test_streaming_get_retries_after_assembly_failure`. max_connections
/// below the node count keeps the PUT fan-out from reaching every node
/// (so a cold getter exists), while min_connections = 3 gives nodes
/// enough ring candidates for the assembly-failure retry to advance to.
async fn setup_assembly_retry_network(name: &str, seed: u64, threshold: usize) -> SimNetwork {
    GlobalRng::set_seed(seed);
    const BASE_EPOCH_MS: u64 = 1577836800000;
    const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
    GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + (seed % RANGE_MS));
    let mut sim = SimNetwork::new(
        name, 1,  // gateways
        12, // nodes
        7,  // ring_max_htl
        3,  // rnd_if_htl_above
        4,  // max_connections (sparse: < node count → cold getter exists)
        3,  // min_connections (≥ 2 retry candidates per node)
        seed,
    )
    .await;
    sim.with_streaming_threshold(threshold);
    // The #4345 assembly-retry test hand-builds a cold-node → holder routing
    // scenario. Placement migration (#4404) moves the contract toward the
    // requester, disrupting that scenario (the injected assembly failure no
    // longer drives the retry path under test). This helper feeds both the
    // phase-1 and phase-2 sims, so pin migration OFF here — the test exercises
    // assembly-retry mechanics, not placement, and must not rely on
    // build-version gating (which flips ON on the v0.2.73 release branch where
    // #4404 ships active).
    sim.disable_placement_migration();
    sim
}

/// Regression test for #4345: a failed stream assembly must consume the
/// GET retry budget and advance to the next candidate instead of
/// synthesizing a client error with the budget untouched.
///
/// Production failure chain: packet loss → FixedRate loss-pause caps the
/// cwnd → the sender aborts on the 3s cwnd-wait timeout → the receiver's
/// stream assembly hits its inactivity timeout → the originator's GET
/// driver (pre-fix) logged a WARN and synthesized "GET succeeded on wire
/// but local store lookup failed" without retrying — one transport
/// hiccup burned the whole GET even though `MAX_RETRIES` was untouched
/// and other candidates could serve the contract.
///
/// Uses the two-phase cold-node isolation strategy from
/// `test_driver_streaming_get_cold_cache`, but on a routable htl=7
/// topology (the HTL=1 variant is not routable — see that test's
/// #[ignore]): phase 1 runs the PUT alone to find the nodes the fan-out
/// missed; phase 2 re-runs the same seed (same ring) and GETs from a
/// cold node with exactly one injected assembly failure armed for this
/// contract key (via `get_assembly_fault_injection`, which mirrors the
/// production failure by leaving the inbound stream orphaned). The
/// driver must advance to another candidate and still deliver the full
/// state.
///
/// Not every fan-out-missed node can route a GET to a holder in a
/// sparse topology (a routing property, not the #4345 assembly-retry
/// property under test), so phase 2 iterates the cold candidates
/// deterministically until one demonstrates the retry path end-to-end.
/// On a regression every candidate fails and the test fails with the
/// per-candidate diagnostics.
#[test]
fn test_streaming_get_retries_after_assembly_failure() {
    use freenet::config::SimulationIdleTimeout;
    use freenet::dev_tool::GET_DRIVER_CALL_COUNT;
    use freenet::dev_tool::get_assembly_fault_injection;
    use std::sync::atomic::Ordering;

    // Root cause of this test's merge-queue flakiness (fix for the #4345
    // follow-up): `run_controlled_simulation` drives Turmoil, whose hosts run
    // their `start_paused(true)` runtimes on THIS thread. The transport's
    // connection idle / keepalive deadline (`PeerConnection::recv`) is checked
    // against `RealTime`, which under a paused runtime is tokio's
    // auto-advancing virtual clock. While a host runs a real (non-mock) WASM
    // contract via `spawn_blocking`, that clock can jump well past the
    // production 120s idle timeout in a single step — and the jumps grow under
    // CPU contention, e.g. the merge-queue runs this alongside four parallel
    // `fdev` simulations that also execute real WASM, so the OS time-slices the
    // blocking work and a starved host's real-time WASM step stretches out.
    // A jump past 120s spuriously tears down otherwise-healthy connections, so
    // the cold-node GET can no longer reach ANY holder. That is exactly the
    // observed failure: under load every candidate reported
    // `retry_path_taken=false` (holder unreachable), whereas in isolation
    // several candidates reach a holder and one demonstrates the retry path.
    //
    // The idle-timeout flag is thread-local and Turmoil runs every host on this
    // thread, so enabling it here (before any `run_controlled_simulation` call)
    // extends the simulation idle timeout to 24h — matching VirtualTime and the
    // documented "120s (RealTime), 24h (VirtualTime/simulation)" contract — and
    // removes the spurious drops without changing what the test verifies (the
    // assembly-failure retry is driven by deterministic fault injection, not by
    // any real connection timing). `run_simulation_direct` enables the same
    // flag for the same reason; we keep it scoped to this test rather than to
    // all of `run_controlled_simulation`, because other controlled-sim tests
    // (e.g. `test_six_peer_contract_lifecycle`) deliberately rely on the
    // production 120s idle timeout to reap stale connections and break if it is
    // extended globally.
    SimulationIdleTimeout::enable();
    // Restore production behaviour on this libtest worker thread when the test
    // returns or unwinds, so the thread-local flag cannot leak into a later
    // test reusing the same thread under plain `cargo test` (nextest isolates
    // per process, so it never leaks there). This makes the "scoped to this
    // test" guarantee above actually hold rather than only mostly hold.
    struct RestoreIdleTimeout;
    impl Drop for RestoreIdleTimeout {
        fn drop(&mut self) {
            freenet::config::SimulationIdleTimeout::disable();
        }
    }
    let _restore_idle_timeout = RestoreIdleTimeout;

    const SEED: u64 = 0xDE1A_4345_0000_0001;
    const NETWORK_NAME_PHASE1: &str = "get-assembly-retry-p1";
    const THRESHOLD: usize = 1024;
    const LARGE_STATE_SIZE: usize = 100 * 1024; // 100KB, above THRESHOLD
    const NODES: usize = 12;

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let contract = SimOperation::create_test_contract(0x45);
    let large_state = SimOperation::create_large_state(LARGE_STATE_SIZE, 0x45);
    let contract_key = contract.key();
    let contract_id = *contract_key.id();

    // Phase 1: run the PUT in isolation to discover the nodes that did
    // NOT receive the state during fan-out.
    let sim1 = rt.block_on(setup_assembly_retry_network(
        NETWORK_NAME_PHASE1,
        SEED,
        THRESHOLD,
    ));
    let phase1_ops = vec![ScheduledOperation::new(
        NodeLabel::gateway(NETWORK_NAME_PHASE1, 0),
        SimOperation::Put {
            contract: contract.clone(),
            state: large_state.clone(),
            subscribe: false,
        },
    )];
    let phase1 = sim1.run_controlled_simulation(
        SEED,
        phase1_ops,
        Duration::from_secs(120),
        Duration::from_secs(30),
    );
    assert!(
        phase1.turmoil_result.is_ok(),
        "Phase 1 (PUT only) should complete: {:?}",
        phase1.turmoil_result.err()
    );
    let cold_nodes: Vec<usize> = (1..=NODES)
        .filter(|i| {
            let label = NodeLabel::node(NETWORK_NAME_PHASE1, *i);
            phase1
                .node_storages
                .get(&label)
                .is_some_and(|s| s.get_stored_state(&contract_key).is_none())
        })
        .collect();
    assert!(
        !cold_nodes.is_empty(),
        "Test precondition: no cold-cache node exists after the PUT in \
         this {NODES}-node sparse topology. All nodes received the state \
         during fan-out. Pick a different SEED or sparser max_connections."
    );

    let mut failures: Vec<String> = Vec::new();
    let mut demonstrated = false;
    // Cap the candidates tried: each iteration is a full phase-2 sim
    // run (~15s), and under a regression EVERY candidate fails — an
    // uncapped loop over up to 11 cold nodes would hit the nextest ci
    // profile's termination timeout and lose the per-candidate
    // diagnostics below. The sim is deterministic per seed; with this
    // seed the first four cold candidates cannot route a GET to a
    // holder at all (their retry exhausts on the wire) and the fifth
    // demonstrates the property, so the cap leaves one spare.
    for (attempt, cold_idx) in cold_nodes.iter().copied().take(6).enumerate() {
        let network_name = format!("get-assembly-retry-p2-{attempt}");

        // (Re-)arm exactly one injected assembly failure for THIS
        // contract key — keying by contract makes this safe under
        // parallel test execution in the same process. `inject_failures`
        // overwrites, so a previous candidate that never streamed leaves
        // no stale budget behind. (ASSEMBLY_RETRY_COUNT itself is
        // process-global; the before/after delta below relies on
        // nextest's process-per-test isolation in CI.)
        get_assembly_fault_injection::inject_failures(contract_key, 1);
        let retries_before =
            get_assembly_fault_injection::ASSEMBLY_RETRY_COUNT.load(Ordering::SeqCst);
        let baseline_calls = GET_DRIVER_CALL_COUNT.load(Ordering::SeqCst);

        let sim2 = rt.block_on(setup_assembly_retry_network(&network_name, SEED, THRESHOLD));
        let cold_node_label = NodeLabel::node(&network_name, cold_idx);
        let phase2_ops = vec![
            ScheduledOperation::new(
                NodeLabel::gateway(&network_name, 0),
                SimOperation::Put {
                    contract: contract.clone(),
                    state: large_state.clone(),
                    subscribe: false,
                },
            ),
            ScheduledOperation::new(
                cold_node_label.clone(),
                SimOperation::Get {
                    contract_id,
                    return_contract_code: false,
                    subscribe: false,
                },
            ),
        ];
        let phase2 = sim2.run_controlled_simulation(
            SEED,
            phase2_ops,
            Duration::from_secs(300),
            Duration::from_secs(120),
        );
        if let Err(e) = &phase2.turmoil_result {
            failures.push(format!("node {cold_idx}: simulation error: {e:?}"));
            continue;
        }

        // Driver-isolation probe: the GET must have gone through the
        // task-per-tx driver (not the local-cache shortcut), otherwise
        // the assembly-retry path was never reachable.
        let driver_calls = GET_DRIVER_CALL_COUNT.load(Ordering::SeqCst);
        if driver_calls == baseline_calls {
            failures.push(format!(
                "node {cold_idx}: GET never reached the driver (locally cached?)"
            ));
            continue;
        }

        let retries_after =
            get_assembly_fault_injection::ASSEMBLY_RETRY_COUNT.load(Ordering::SeqCst);
        let stored = phase2
            .node_storages
            .get(&cold_node_label)
            .and_then(|s| s.get_stored_state(&contract_key));

        match (retries_after > retries_before, stored) {
            (true, Some(state)) => {
                // The driver took the assembly-failure retry path AND
                // the retry delivered the full, correct state.
                let stored_bytes: Vec<u8> = state.as_ref().to_vec();
                assert_eq!(
                    stored_bytes, large_state,
                    "Stored state bytes should match the original 100KB \
                     state after the assembly-failure retry"
                );
                demonstrated = true;
                break;
            }
            (took_retry, stored) => {
                failures.push(format!(
                    "node {cold_idx}: retry_path_taken={took_retry} \
                     state_present={} (streaming GET likely never reached \
                     a holder from this candidate, or the retry burned \
                     the GET — #4345 regression if this repeats for every \
                     candidate)",
                    stored.is_some()
                ));
            }
        }
    }

    assert!(
        demonstrated,
        "No cold candidate demonstrated the #4345 assembly-failure retry \
         path (injected failure → driver.advance() → fresh attempt → \
         state delivered). Per-candidate outcomes: {failures:#?}"
    );
}
