repro.rs

use std::time::Duration;

use discord_cassandra_cpp::{session_scope::Unbound, *};
use tokio::{join, time::sleep};

async fn create_unbound_session() -> Result<Session<Unbound>> {
    let contact_points = "127.0.0.1";
    let mut cluster = Cluster::default();
    cluster.set_contact_points(contact_points)?;
    cluster.set_load_balance_round_robin();

    cluster.connect().await
}

async fn create_session() -> Result<Session> {
    let contact_points = "127.0.0.1";
    let mut cluster = Cluster::default();
    cluster.set_contact_points(contact_points)?;
    cluster.set_load_balance_round_robin();
    cluster.connect_keyspace("discord_udt_bug_repro").await
}

async fn setup_db() -> Result<Session<Unbound>> {
    let session = create_unbound_session().await?;

    session
        .execute("DROP KEYSPACE IF EXISTS discord_udt_bug_repro")
        .await?;

    session.execute("CREATE KEYSPACE discord_udt_bug_repro WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1}")
        .await?;

    session
        .execute("CREATE TYPE discord_udt_bug_repro.basic_udt (foo text, bar text)")
        .await?;

    session.execute("CREATE TABLE discord_udt_bug_repro.basic_table (id int, udt_data frozen<basic_udt>, PRIMARY KEY (id))")
        .await?;

    Ok(session)
}

#[tokio::main]
async fn main() -> Result<()> {
    let unbound_session = setup_db().await?;
    let session = create_session().await?;
    let prepared_statement = session
        .prepare("INSERT INTO basic_table (id, udt_data) VALUES (?, ?)")
        .await?;

    let mut loopy_boi = tokio::task::spawn(async move {
        // lets do some insertions, cos why not...

        let mut count = 0;

        loop {
            let mut statement = prepared_statement.bind();
            statement
                .bind_int32_by_name("id", count)
                .expect("failed to bind id");

            let mut user_type = session
                .get_schema_meta()
                .get_keyspace_by_name(session.keyspace())
                .user_type_by_name("basic_udt")
                .expect("failed to get user type")
                .new_user_type();

            user_type
                .set_string_by_name("foo", &format!("foo {}", count))
                .expect("failed to set basic_udt.foo");

            user_type
                .set_string_by_name("bar", &format!("bar {}", count))
                .expect("failed to set basic_udt.bar");

            statement
                .bind_user_type_by_name("udt_data", &user_type)
                .expect("failed to bind udt_data"); // <-- failure occurs here.

            statement
                .execute()
                .await
                .expect("failed to insert into basic table");

            println!("inserted row {}", count);

            count += 1;

            sleep(Duration::from_millis(500)).await;
        }
    });

    sleep(Duration::from_secs(1)).await;

    println!("altering udt");
    unbound_session
        .execute("ALTER TYPE discord_udt_bug_repro.basic_udt ADD baz blob")
        .await?;
    println!("altered udt!");

    let (_, result) = join!(sleep(Duration::from_secs(2)), &mut loopy_boi);

    result.expect("loopy boi crashed");

    loopy_boi.abort();

    Ok(())
}