Function webhook_concurrent_actions

src/environmentd/tests/server.rs:3599–3749

()

#[cfg_attr(miri, ignore)] // too slow
#[allow(clippy::disallowed_methods)]
async fn webhook_concurrent_actions() {
    let server = test_util::TestHarness::default().start().await;
    let client = server.connect().await.unwrap();

    #[derive(Debug, PartialEq, Eq, Deserialize, Serialize)]
    struct WebhookEvent {
        ts: u128,
        name: String,
        thread: usize,
    }

    // Create a webhook source.
    let src_name = "webhook_json";
    client
        .execute(
            "CREATE CLUSTER webhook_cluster_concurrent REPLICAS (r1 (SIZE 'scale=1,workers=1'));",
            &[],
        )
        .await
        .expect("failed to create cluster");
    client
        .execute(
            &format!("CREATE SOURCE {src_name} IN CLUSTER webhook_cluster_concurrent FROM WEBHOOK BODY FORMAT JSON"),
            &[],
        )
        .await
        .expect("failed to create source");

    fn now() -> u128 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("time went backwards")
            .as_nanos()
    }

    // Flag that lets us shutdown our threads.
    let keep_sending = Arc::new(AtomicBool::new(true));
    // Track how many requests were resolved before we dropped the collection.
    let num_requests_before_drop = Arc::new(AtomicUsize::new(0));

    // Spin up tasks that will contiously push data to the webhook.
    let keep_sending_ = Arc::clone(&keep_sending);
    let num_requests_before_drop_ = Arc::clone(&num_requests_before_drop);
    let addr = server.http_local_addr();

    let poster = mz_ore::task::spawn(|| "webhook_concurrent_actions-poster", async move {
        let mut i = 0;

        let http_client = reqwest::Client::new();
        let mut tasks = Vec::with_capacity(500);

        // Keep sending events until we're told to stop.
        while keep_sending_.load(std::sync::atomic::Ordering::Relaxed) {
            let webhook_url = format!(
                "http://{}/api/webhook/materialize/public/webhook_json",
                addr,
            );