Function test_github_20262

src/environmentd/tests/server.rs:3337–3398

()

#[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `epoll_wait` on OS `linux`
#[allow(clippy::disallowed_methods)]
fn test_github_20262() {
    let server = test_util::TestHarness::default().start_blocking();
    let mut client = server.connect(postgres::NoTls).unwrap();
    client.batch_execute("CREATE TABLE t (i INT);").unwrap();

    let mut cancel = || {
        // Wait for the subscription to start.
        let conn_id = Retry::default()
            .retry(|_| {
                let conn_id: String = client
                    .query_one(
                        "SELECT s.connection_id::text FROM mz_internal.mz_subscriptions b JOIN mz_internal.mz_sessions s ON s.id = b.session_id",
                        &[],
                    )?
                    .get(0);
                Ok::<_, postgres::Error>(conn_id)
            })
            .unwrap();
        client
            .query_one(&format!("SELECT pg_cancel_backend({conn_id})"), &[])
            .unwrap();
    };

    let subscribe: serde_json::Value =
        serde_json::from_str(r#"{"queries":[{"query":"SUBSCRIBE t"}]}"#).unwrap();
    let subscribe = subscribe.to_string();
    let commit: serde_json::Value =
        serde_json::from_str(r#"{"queries":[{"query":"COMMIT"}]}"#).unwrap();
    let commit = commit.to_string();
    let select: serde_json::Value =
        serde_json::from_str(r#"{"queries":[{"query":"SELECT 1"}]}"#).unwrap();
    let select = select.to_string();

    let (mut ws, _resp) = tungstenite::connect(server.ws_addr()).unwrap();
    test_util::auth_with_ws(&mut ws, BTreeMap::default()).unwrap();
    ws.send(Message::Text(subscribe.into())).unwrap();
    cancel();
    ws.send(Message::Text(commit.into())).unwrap();
    ws.send(Message::Text(select.into())).unwrap();

    let mut expect = VecDeque::from([
        r#"{"type":"CommandStarting","payload":{"has_rows":true,"is_streaming":true}}"#.to_string(),
        r#"{"type":"Rows","payload":{"columns":[{"name":"mz_timestamp","type_oid":1700,"type_len":-1,"type_mod":2555908},{"name":"mz_diff","type_oid":20,"type_len":8,"type_mod":-1},{"name":"i","type_oid":23,"type_len":4,"type_mod":-1}]}}"#.to_string(),
        r#"{"type":"Error","payload":{"message":"canceling statement due to user request","code":"57014"}}"#.to_string(),
        r#"{"type":"ReadyForQuery","payload":"I"}"#.to_string(),
        r#"{"type":"Notice","payload":{"message":"there is no transaction in progress","code":"25P01","severity":"warning"}}"#.to_string(),
        r#"{"type":"CommandStarting","payload":{"has_rows":false,"is_streaming":false}}"#.to_string(),
        r#"{"type":"CommandComplete","payload":"COMMIT"}"#.to_string(),
        r#"{"type":"ReadyForQuery","payload":"I"}"#.to_string(),
        r#"{"type":"CommandStarting","payload":{"has_rows":true,"is_streaming":false}}"#.to_string(),
        format!(r#"{{"type":"Rows","payload":{{"columns":[{{"name":"{UNKNOWN_COLUMN_NAME}","type_oid":23,"type_len":4,"type_mod":-1}}]}}}}"#),
        r#"{"type":"Row","payload":["1"]}"#.to_string(),
        r#"{"type":"CommandComplete","payload":"SELECT 1"}"#.to_string(),
        r#"{"type":"ReadyForQuery","payload":"I"}"#.to_string(),
    ]);
    while !expect.is_empty() {
        if let Message::Text(text) = ws.read().unwrap() {
            let next = expect.pop_front().unwrap();