Repository: tikv/tikv
File: tests/failpoints/cases/test_merge.rs (Rust)

Code before:
    };
    let (tx, rx) = std::sync::mpsc::channel();
    let callback = Callback::read(Box::new(move |req| {
        tx.send(req).unwrap();
    }));
    let cmd_req = RaftCommand::new(cmd, callback);
    raft_router.send_raft_command(cmd_req).unwrap();

    // send a casual msg that can trigger panic after peer fms closed.
    raft_router
        .send_casual_msg(1, CasualMessage::ForceCompactRaftLogs)
        .unwrap();

    fail::remove("pause_on_peer_collect_message");

    // wait some time for merge finish.
    std::thread::sleep(Duration::from_secs(1));
    must_get_equal(&cluster.get_engine(1), b"k3", b"v3");

    let resp = rx.recv().unwrap();
    assert!(resp.response.get_header().has_error());
}

// This case tests following scenario:
// When `max_apply_unpersisted_log_limit > 0`, raft leader's applied_index can
// be higher than committed_index. When the leader propose a merge, in the
// CatchUpLogs phase, it should ensure the all peers can be recovered from the
// log even if any peer is restarted in which case it can lose some unpersisted
// logs. For more details, see: https://github.com/tikv/tikv/issues/18129
#[test]
fn test_node_merge_with_apply_ahead_of_persist() {
    let mut cluster = new_node_cluster(0, 3);
    configure_for_merge(&mut cluster.cfg);
    cluster.cfg.raft_store.cmd_batch_concurrent_ready_max_count = 32;
    cluster.cfg.raft_store.store_io_pool_size = 1;
    cluster.cfg.raft_store.max_apply_unpersisted_log_limit = 1024;

    cluster.run();
    let pd_client = Arc::clone(&cluster.pd_client);
    pd_client.disable_default_operator();
    cluster.must_transfer_leader(1, new_peer(1, 1));

    cluster.must_put(b"k1", b"v1");
    cluster.must_put(b"k3", b"v3");

    let region = cluster.get_region(b"k1");
    cluster.must_split(&region, b"k2");
    let left = cluster.get_region(b"k1");
    let right = cluster.get_region(b"k3");

    cluster.must_put(b"k1", b"v2");

Review comment (suggestion):
> raft leader's applied_index can be higher than committed_index

Would you mind explaining how `applied_index` is related to the problem being solved? From the fix above, it looks like only match_index and committed_index are related. 

In other words, I'd like to understand why this issue didn't happen when "apply before persistence" was turned off. Thx!

Code after:
    };
    let (tx, rx) = std::sync::mpsc::channel();
    let callback = Callback::read(Box::new(move |req| {
        tx.send(req).unwrap();
    }));
    let cmd_req = RaftCommand::new(cmd, callback);
    raft_router.send_raft_command(cmd_req).unwrap();

    // send a casual msg that can trigger panic after peer fms closed.
    raft_router
        .send_casual_msg(1, CasualMessage::ForceCompactRaftLogs)
        .unwrap();

    fail::remove("pause_on_peer_collect_message");

    // wait some time for merge finish.
    std::thread::sleep(Duration::from_secs(1));
    must_get_equal(&cluster.get_engine(1), b"k3", b"v3");

    let resp = rx.recv().unwrap();
    assert!(resp.response.get_header().has_error());
}

// This case tests following scenario:
// When the disk IO is slow or block and PrepareMerge is proposed, it is
// possible that the leader's committed index is higher than its persisted
// index. In this case, we should still ensure that all peers can recover
// necessary raft logs from the CatchUpLogs phase.
// For more information, see: https://github.com/tikv/tikv/issues/18129
#[test]
fn test_node_merge_with_apply_ahead_of_persist() {
    let mut cluster = new_node_cluster(0, 3);
    configure_for_merge(&mut cluster.cfg);
    cluster.cfg.raft_store.cmd_batch_concurrent_ready_max_count = 32;
    cluster.cfg.raft_store.store_io_pool_size = 1;
    // even if "early apply" is disabled, the raft committed index can still be
    // higher than persisted/matched index.
    cluster.cfg.raft_store.max_apply_unpersisted_log_limit = 0;

    cluster.run();
    let pd_client = Arc::clone(&cluster.pd_client);
    pd_client.disable_default_operator();
    cluster.must_transfer_leader(1, new_peer(1, 1));

    cluster.must_put(b"k1", b"v1");
    cluster.must_put(b"k3", b"v1");

    let region = cluster.get_region(b"k1");
    cluster.must_split(&region, b"k2");
    let left = cluster.get_region(b"k1");
    let right = cluster.get_region(b"k3");