use crate::common::{
    create_test_metrics, produce_cluster_node, produce_cluster_node_with_metrics, start_etcd,
    TestMetricsCollector,
};
use photon_etcd_cluster::{ClusterEvent, ServiceDiscovery};
use serial_test::serial;
use std::sync::Once;
use tokio::{sync::broadcast, time::timeout};

mod common;

static TRACING_INIT: Once = Once::new();
static DEFAULT_GROUP: &str = "default";

pub fn init_tracing() {
    TRACING_INIT.call_once(|| {
        tracing_subscriber::fmt()
            .with_env_filter("info,observability=info")
            .with_test_writer()
            .init();
    });
}

#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
#[serial]
async fn single_worker_elects_itself_and_monitor_sees_it() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // --- spawn cluster node ---
    let (shutdown_tx, shutdown_rv) = broadcast::channel::<()>(8);
    let node_id = format!("node-{}", uuid::Uuid::new_v4());
    let node = produce_cluster_node(endpoints.clone(), &node_id);
    let node_task = {
        let n = node.clone();
        let mut rx = shutdown_rv.resubscribe();
        tokio::spawn(async move { n.run(&mut rx).await })
    };

    // --- spawn service discovery ---
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_task = tokio::spawn(async move { disc.run(None).await });

    // Wait until discovery reports the node and the same node is leader

    timeout(std::time::Duration::from_secs(3), async {
        loop {
            if let Some(leader) = discovery.leader() {
                if leader.id == node_id {
                    assert_eq!(discovery.nodes().len(), 1);
                    break;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("condition not met in time");

    // --- graceful shutdown ---
    shutdown_tx.send(()).unwrap();
    node_task.await.unwrap().unwrap();
    disc_task.abort();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[serial]
async fn leader_reelection_on_failure() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    let (n1_s_tx, mut n1_s_rv) = broadcast::channel::<()>(1);
    let t1 = {
        let n = produce_cluster_node(endpoints.clone(), "n1");
        tokio::spawn(async move { n.run(&mut n1_s_rv).await })
    };

    let (n2_s_tx, mut n2_s_rv) = broadcast::channel::<()>(1);
    let t2 = {
        let n = produce_cluster_node(endpoints.clone(), "n2");
        tokio::spawn(async move { n.run(&mut n2_s_rv).await })
    };

    // service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_t = tokio::spawn(async move { disc.run(None).await });

    // wait for first leader
    timeout(std::time::Duration::from_secs(3), async {
        loop {
            match discovery.leader() {
                Some(l) => {
                    if l.id == "n1" || l.id == "n2" {
                        assert_eq!(discovery.nodes().len(), 2);
                        break;
                    }
                }
                _ => tokio::time::sleep(std::time::Duration::from_millis(50)).await,
            }
        }
    })
    .await
    .expect("election timeout");
    let first_leader_id = discovery.leader().expect("have leader").id;
    assert!(first_leader_id == "n1" || first_leader_id == "n2");

    // kill that node
    if first_leader_id == "n1" {
        n1_s_tx.send(()).unwrap();
        t1.await.unwrap().unwrap();
    } else {
        n2_s_tx.send(()).unwrap();
        t2.await.unwrap().unwrap();
    }

    // within TTL another leader should appear
    timeout(std::time::Duration::from_secs(2), async {
        loop {
            match discovery.leader() {
                Some(l) => {
                    if l.id != first_leader_id {
                        assert_eq!(discovery.nodes().len(), 1);
                        break;
                    }
                }
                _ => tokio::time::sleep(std::time::Duration::from_millis(50)).await,
            }
        }
    })
    .await
    .expect("new leader not elected");

    // shutdown remaining tasks
    n1_s_tx.send(()).ok();
    n2_s_tx.send(()).ok();
    disc_t.abort();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[serial]
async fn nodes_reconnect_on_network_failure() {
    init_tracing();

    let mut etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    let (n1_s_tx, mut n1_s_rv) = broadcast::channel::<()>(1);
    let t1 = {
        let n = produce_cluster_node(endpoints.clone(), "n1");
        tokio::spawn(async move { n.run(&mut n1_s_rv).await })
    };

    let (n2_s_tx, mut n2_s_rv) = broadcast::channel::<()>(1);
    let t2 = {
        let n = produce_cluster_node(endpoints.clone(), "n2");
        tokio::spawn(async move { n.run(&mut n2_s_rv).await })
    };

    // service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_t = tokio::spawn(async move { disc.run(None).await });

    tokio::time::sleep(std::time::Duration::from_secs(2)).await;

    etcd.restart().await;

    tokio::time::sleep(std::time::Duration::from_secs(10)).await;

    // wait for nodes to re-connect
    timeout(std::time::Duration::from_secs(10), async {
        loop {
            if discovery.nodes().len() == 2 {
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("election timeout");
    let first_leader_id = discovery.leader().expect("have leader").id;
    assert!(first_leader_id == "n1" || first_leader_id == "n2");

    // shutdown
    n1_s_tx.send(()).ok();
    n2_s_tx.send(()).ok();
    t1.await.unwrap().unwrap();
    t2.await.unwrap().unwrap();
    disc_t.abort();
}

/// Test that a node is immediately removed from etcd when it gracefully shuts down.
/// This is critical for service discovery - nodes should not linger in etcd after shutdown.
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
#[serial]
async fn node_immediately_removed_on_graceful_shutdown() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // Spawn cluster node
    let (shutdown_tx, mut shutdown_rx) = broadcast::channel::<()>(8);
    let node_id = format!("node-{}", uuid::Uuid::new_v4());
    let node = produce_cluster_node(endpoints.clone(), &node_id);
    let node_task = {
        let n = node.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx).await })
    };

    // Spawn service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_task = tokio::spawn(async move { disc.run(None).await });

    // Wait until discovery sees the node
    timeout(std::time::Duration::from_secs(5), async {
        loop {
            if discovery.nodes().iter().any(|n| n.id == node_id) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("Node not discovered in time");

    assert_eq!(discovery.nodes().len(), 1, "Should have exactly 1 node");

    // Record the time before shutdown
    let shutdown_start = std::time::Instant::now();

    // Send shutdown signal and wait for task to complete
    shutdown_tx.send(()).unwrap();
    node_task.await.unwrap().unwrap();

    // Verify node is removed IMMEDIATELY (within 500ms, not waiting for TTL which is 3s)
    // The node should be removed via explicit delete, not lease expiry
    timeout(std::time::Duration::from_millis(500), async {
        loop {
            if discovery.nodes().is_empty() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("Node was not immediately removed from etcd on shutdown");

    let removal_time = shutdown_start.elapsed();
    tracing::info!(
        "Node removed in {:?} (should be < 500ms, TTL is 3s)",
        removal_time
    );

    // Verify removal happened quickly (not waiting for TTL)
    assert!(
        removal_time.as_millis() < 500,
        "Node removal took {:?}, should be immediate (< 500ms), not TTL-based",
        removal_time
    );

    disc_task.abort();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 10)]
#[serial]
async fn many_nodes_leader_reelection() {
    init_tracing();

    let etcd = common::start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // Spawn 20 cluster nodes
    let mut node_shutdown_txs = Vec::new();
    let mut node_handles = Vec::new();
    for i in 0..20 {
        let node_id = format!("node-{}", i);
        let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1);
        let node = common::produce_cluster_node(endpoints.clone(), &node_id);
        let handle = tokio::spawn(async move {
            node.run(&mut shutdown_rx).await.unwrap();
        });
        node_shutdown_txs.push(shutdown_tx);
        node_handles.push(handle);
    }

    // Spawn service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let discovery_handle = tokio::spawn(async move {
        disc.run(None).await.unwrap();
    });

    // Wait for initial setup: all nodes registered and a leader elected
    timeout(std::time::Duration::from_secs(30), async {
        loop {
            let nodes = discovery.nodes();
            let leader = discovery.leader();
            if nodes.len() == 20 && leader.is_some() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("Initial leader election failed");

    // Kill nodes one by one and check leader existence
    for (i, shutdown_tx) in node_shutdown_txs.into_iter().enumerate() {
        shutdown_tx.send(()).unwrap();

        let target_nodes = 20 - (i + 1);
        // Wait for node count to decrease
        timeout(std::time::Duration::from_secs(10), async {
            loop {
                if discovery.nodes().len() == target_nodes {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("Node count did not decrease in time");

        if target_nodes > 0 {
            // Ensure a valid leader exists
            timeout(std::time::Duration::from_secs(10), async {
                loop {
                    if let Some(leader) = discovery.leader() {
                        if discovery.nodes().iter().any(|n| n.id == leader.id) {
                            break;
                        }
                    }
                    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                }
            })
            .await
            .expect("Leader not present after kill");
        } else {
            assert!(discovery.leader().is_none());
        }
    }

    // Cleanup service discovery
    discovery_handle.abort();

    // Ensure all nodes have finished (though they should be done by now)
    for handle in node_handles {
        handle.await.unwrap();
    }
}

/// Test that node metadata is stored in etcd and correctly retrieved via ServiceDiscovery.
/// This verifies the end-to-end flow of metadata from ClusterNode to etcd to ServiceDiscovery.
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
#[serial]
async fn node_metadata_stored_and_retrieved() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // Create a node with specific test metrics
    let test_metrics = create_test_metrics(45.5, 62.0, 10);
    let collector = TestMetricsCollector::new(test_metrics.clone());

    let (shutdown_tx, mut shutdown_rx) = broadcast::channel::<()>(8);
    let node_id = format!("node-metrics-{}", uuid::Uuid::new_v4());

    // Create node with metrics (no periodic updates - just initial registration)
    let node = produce_cluster_node_with_metrics(
        endpoints.clone(),
        &node_id,
        collector,
        0, // No periodic updates for this test
    );

    let node_task = {
        let n = node.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx).await })
    };

    // Spawn service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_task = tokio::spawn(async move { disc.run(None).await });

    // Wait until discovery sees the node with metadata
    let retrieved_metadata = timeout(std::time::Duration::from_secs(5), async {
        loop {
            let nodes = discovery.nodes();
            if let Some(node) = nodes.iter().find(|n| n.id == node_id) {
                // Check if metadata is present (not null)
                if !node.metadata.is_null() {
                    return node.metadata.clone();
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("Node metadata not retrieved in time");

    // Verify the metadata matches what we set
    assert_eq!(
        retrieved_metadata["cpu_usage_percent"].as_f64(),
        Some(45.5),
        "CPU usage should match"
    );
    assert_eq!(
        retrieved_metadata["memory_usage_percent"].as_f64(),
        Some(62.0),
        "Memory usage should match"
    );
    assert_eq!(
        retrieved_metadata["queue_depth"].as_u64(),
        Some(10),
        "Queue depth should match"
    );
    assert_eq!(
        retrieved_metadata["custom_field"].as_str(),
        Some("test_value"),
        "Custom field should match"
    );

    tracing::info!(
        "Successfully retrieved metadata from etcd: {:?}",
        retrieved_metadata
    );

    // Graceful shutdown
    shutdown_tx.send(()).unwrap();
    node_task.await.unwrap().unwrap();
    disc_task.abort();
}

/// Test that metadata updates are propagated to ServiceDiscovery via NodeUpdated events.
/// This verifies the periodic metrics update task works correctly.
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
#[serial]
async fn node_metadata_updates_propagated() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // Create a node with initial metrics
    let initial_metrics = create_test_metrics(30.0, 40.0, 5);
    let collector = TestMetricsCollector::new(initial_metrics);
    let metrics_handle = collector.metrics_handle();

    let (shutdown_tx, mut shutdown_rx) = broadcast::channel::<()>(8);
    let node_id = format!("node-update-{}", uuid::Uuid::new_v4());

    // Create node with metrics and periodic updates every 2 seconds
    let node = produce_cluster_node_with_metrics(
        endpoints.clone(),
        &node_id,
        collector,
        2, // Update every 2 seconds
    );

    let node_task = {
        let n = node.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx).await })
    };

    // Spawn service discovery with event subscription
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let mut events = discovery.subscribe();
    let disc = discovery.clone();
    let disc_task = tokio::spawn(async move { disc.run(None).await });

    // Wait for initial node registration with metadata
    timeout(std::time::Duration::from_secs(5), async {
        loop {
            let nodes = discovery.nodes();
            if let Some(node) = nodes.iter().find(|n| n.id == node_id) {
                if !node.metadata.is_null() {
                    return;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("Initial node not found");

    // Verify initial metrics
    let initial_node = discovery
        .nodes()
        .iter()
        .find(|n| n.id == node_id)
        .cloned()
        .expect("Node should exist");
    assert_eq!(
        initial_node.metadata["cpu_usage_percent"].as_f64(),
        Some(30.0)
    );
    assert_eq!(
        initial_node.metadata["memory_usage_percent"].as_f64(),
        Some(40.0)
    );

    // Update the metrics in the collector (simulating changing system load)
    let updated_metrics = create_test_metrics(75.0, 85.0, 50);
    {
        *metrics_handle.write().unwrap() = updated_metrics.clone();
    }

    tracing::info!("Metrics updated in collector, waiting for propagation...");

    // Wait for NodeUpdated event with new metadata
    let update_event = timeout(std::time::Duration::from_secs(10), async {
        loop {
            match events.recv().await {
                Ok(ClusterEvent::NodeUpdated { old, new }) => {
                    if new.id == node_id
                        && new.metadata["cpu_usage_percent"].as_f64() == Some(75.0)
                    {
                        return (old, new);
                    }
                }
                Ok(_) => continue,
                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                Err(_) => panic!("Event channel closed"),
            }
        }
    })
    .await
    .expect("NodeUpdated event with new metadata not received");

    let (old_node, new_node) = update_event;
    tracing::info!(
        "Received NodeUpdated event: old CPU={:?}, new CPU={:?}",
        old_node.metadata["cpu_usage_percent"],
        new_node.metadata["cpu_usage_percent"]
    );

    // Verify old values
    assert_eq!(
        old_node.metadata["cpu_usage_percent"].as_f64(),
        Some(30.0),
        "Old CPU should be 30.0"
    );
    assert_eq!(
        old_node.metadata["memory_usage_percent"].as_f64(),
        Some(40.0),
        "Old memory should be 40.0"
    );

    // Verify new values
    assert_eq!(
        new_node.metadata["cpu_usage_percent"].as_f64(),
        Some(75.0),
        "New CPU should be 75.0"
    );
    assert_eq!(
        new_node.metadata["memory_usage_percent"].as_f64(),
        Some(85.0),
        "New memory should be 85.0"
    );
    assert_eq!(
        new_node.metadata["queue_depth"].as_u64(),
        Some(50),
        "New queue depth should be 50"
    );

    // Also verify via direct nodes() access
    let current_node = discovery
        .nodes()
        .iter()
        .find(|n| n.id == node_id)
        .cloned()
        .expect("Node should still exist");
    assert_eq!(
        current_node.metadata["cpu_usage_percent"].as_f64(),
        Some(75.0),
        "Direct access should show new CPU"
    );

    // Graceful shutdown
    shutdown_tx.send(()).unwrap();
    node_task.await.unwrap().unwrap();
    disc_task.abort();
}

/// Test that multiple nodes with different metadata are correctly tracked.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[serial]
async fn multiple_nodes_with_different_metadata() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // Create three nodes with different metrics
    let metrics1 = create_test_metrics(10.0, 20.0, 1);
    let metrics2 = create_test_metrics(50.0, 60.0, 10);
    let metrics3 = create_test_metrics(90.0, 95.0, 100);

    let collector1 = TestMetricsCollector::new(metrics1);
    let collector2 = TestMetricsCollector::new(metrics2);
    let collector3 = TestMetricsCollector::new(metrics3);

    let (shutdown_tx1, mut shutdown_rx1) = broadcast::channel::<()>(1);
    let (shutdown_tx2, mut shutdown_rx2) = broadcast::channel::<()>(1);
    let (shutdown_tx3, mut shutdown_rx3) = broadcast::channel::<()>(1);

    let node1 = produce_cluster_node_with_metrics(endpoints.clone(), "node-m1", collector1, 0);
    let node2 = produce_cluster_node_with_metrics(endpoints.clone(), "node-m2", collector2, 0);
    let node3 = produce_cluster_node_with_metrics(endpoints.clone(), "node-m3", collector3, 0);

    let t1 = {
        let n = node1.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx1).await })
    };
    let t2 = {
        let n = node2.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx2).await })
    };
    let t3 = {
        let n = node3.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx3).await })
    };

    // Spawn service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_task = tokio::spawn(async move { disc.run(None).await });

    // Wait for all three nodes with metadata
    timeout(std::time::Duration::from_secs(10), async {
        loop {
            let nodes = discovery.nodes();
            let nodes_with_metadata: Vec<_> = nodes
                .iter()
                .filter(|n| !n.metadata.is_null())
                .collect();
            if nodes_with_metadata.len() == 3 {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("Not all nodes registered with metadata");

    // Verify each node has correct metadata
    let nodes = discovery.nodes();

    let n1 = nodes.iter().find(|n| n.id == "node-m1").expect("node-m1");
    assert_eq!(n1.metadata["cpu_usage_percent"].as_f64(), Some(10.0));
    assert_eq!(n1.metadata["queue_depth"].as_u64(), Some(1));

    let n2 = nodes.iter().find(|n| n.id == "node-m2").expect("node-m2");
    assert_eq!(n2.metadata["cpu_usage_percent"].as_f64(), Some(50.0));
    assert_eq!(n2.metadata["queue_depth"].as_u64(), Some(10));

    let n3 = nodes.iter().find(|n| n.id == "node-m3").expect("node-m3");
    assert_eq!(n3.metadata["cpu_usage_percent"].as_f64(), Some(90.0));
    assert_eq!(n3.metadata["queue_depth"].as_u64(), Some(100));

    tracing::info!("All three nodes have correct metadata");

    // Graceful shutdown
    shutdown_tx1.send(()).unwrap();
    shutdown_tx2.send(()).unwrap();
    shutdown_tx3.send(()).unwrap();
    t1.await.unwrap().unwrap();
    t2.await.unwrap().unwrap();
    t3.await.unwrap().unwrap();
    disc_task.abort();
}

/// Test backward compatibility: nodes without metadata work alongside nodes with metadata.
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
#[serial]
async fn backward_compatibility_nodes_without_metadata() {
    init_tracing();

    let etcd = start_etcd().await;
    let endpoints = vec![etcd.endpoint.clone()];

    // Create one node WITH metadata
    let metrics = create_test_metrics(50.0, 60.0, 10);
    let collector = TestMetricsCollector::new(metrics);

    let (shutdown_tx1, mut shutdown_rx1) = broadcast::channel::<()>(1);
    let (shutdown_tx2, mut shutdown_rx2) = broadcast::channel::<()>(1);

    // Node with metadata
    let node_with_meta =
        produce_cluster_node_with_metrics(endpoints.clone(), "node-with-meta", collector, 0);

    // Node without metadata (using the old-style constructor)
    let node_without_meta = produce_cluster_node(endpoints.clone(), "node-without-meta");

    let t1 = {
        let n = node_with_meta.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx1).await })
    };
    let t2 = {
        let n = node_without_meta.clone();
        tokio::spawn(async move { n.run(&mut shutdown_rx2).await })
    };

    // Spawn service discovery
    let discovery = ServiceDiscovery::new(endpoints, DEFAULT_GROUP.into());
    let disc = discovery.clone();
    let disc_task = tokio::spawn(async move { disc.run(None).await });

    // Wait for both nodes
    timeout(std::time::Duration::from_secs(5), async {
        loop {
            if discovery.nodes().len() == 2 {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("Both nodes not registered");

    let nodes = discovery.nodes();

    // Node with metadata should have it
    let with_meta = nodes
        .iter()
        .find(|n| n.id == "node-with-meta")
        .expect("node-with-meta");
    assert!(
        !with_meta.metadata.is_null(),
        "Node with metadata should have non-null metadata"
    );
    assert_eq!(with_meta.metadata["cpu_usage_percent"].as_f64(), Some(50.0));

    // Node without metadata should have null metadata
    let without_meta = nodes
        .iter()
        .find(|n| n.id == "node-without-meta")
        .expect("node-without-meta");
    assert!(
        without_meta.metadata.is_null(),
        "Node without metadata should have null metadata"
    );

    tracing::info!("Backward compatibility verified: both node types coexist");

    // Graceful shutdown
    shutdown_tx1.send(()).unwrap();
    shutdown_tx2.send(()).unwrap();
    t1.await.unwrap().unwrap();
    t2.await.unwrap().unwrap();
    disc_task.abort();
}
