You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: backfill.rs
    path of file: ./repos/risingwave/src/tests/simulation/tests/integration_tests/recovery
    
    the code of the file until where you have to start completion: // Copyright 2024 RisingWave Labs
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the L
    
async fn test_snapshot_mv() -> Result<()> {
    let mut cluster = Cluster::start(Configuration::for_scale()).await?;
    let mut session = cluster.start_session();

    session.run(ROOT_TABLE_CREATE).await?;
    session.run(MV1).await?;

    test_no_backfill_state(&mut session).await?;

    session
        .run(&format!(
            "insert into t1 values {}",
            (1..=10).map(|x| format!("({x})")).join(",")
        ))
        .await?;

    session.run("flush").await?;

    // v1 > 5, result is [6, 7, 8, 9, 10]
    session
        .run("select count(*) from m1")
        .await?
        .assert_result_eq("5");

    // After startup with no backfill, with data inserted after, should be NO_BACKFILL state.
    test_no_backfill_state(&mut session).await?;

    let fragment = cluster
        .locate_one_fragment([
            identity_contains("materialize"),
            no_identity_contains("StreamTableScan"),
        ])
        .await?;

    let id = fragment.id();

    cluster.reschedule(format!("{id}-[1,2,3,4,5]")).await?;
    sleep(Duration::from_secs(3)).await;

    // Before complete recovery should be NO_BACKFILL state
    test_no_backfill_state(&mut session).await?;

    cluster.reschedule(format!("{id}+[1,2,3,4,5]")).await?;
    sleep(Duration::from_secs(3)).await;

    // After recovery should be NO_BACKFILL state
    test_no_backfill_state(&mut session).await?;

    Ok(())
}