// Copyright 2026 PingCAP, Inc.

use std::sync::Arc;

use pd_client::PdClient;
use test_rfstore::*;
use tikv_util::store::find_peer;

#[test]
fn test_rfstore_async_io_commit_without_leader_persist() {
    let mut cluster = new_node_cluster(2, 3);
    cluster.cfg.raft_store.max_apply_unpersisted_log_limit = 0;
    let pd_client = Arc::clone(&cluster.pd_client);
    pd_client.disable_default_operator();

    cluster.run();

    let region = pd_client.get_region(b"k1").unwrap();
    let peer_1 = find_peer(&region, 1).cloned().unwrap();

    cluster.must_put(b"k1", b"v1");
    cluster.must_transfer_leader(region.get_id(), peer_1);

    let raft_before_save_on_store_1_fp = "rfstore_before_save_on_store_1";
    fail::cfg(raft_before_save_on_store_1_fp, "pause").unwrap();

    let pending_writes = (2..10)
        .map(|i| {
            cluster
                .async_put(format!("k{}", i).as_bytes(), b"v1")
                .unwrap()
        })
        .collect::<Vec<_>>();
    let _ = tikv_util::future::block_on_timeout(
        futures::future::join_all(pending_writes),
        std::time::Duration::from_secs(5),
    );

    // Although leader cannot persist entries, followers can still commit them.
    must_get_equal(&cluster.get_engine(2), region.id, b"k9", b"v1");
    must_get_equal(&cluster.get_engine(3), region.id, b"k9", b"v1");
    // Entries must still wait for persistence before local apply on the leader.
    must_get_none(&cluster.get_engine(1), region.id, b"k9");

    fail::remove(raft_before_save_on_store_1_fp);
    must_get_equal(&cluster.get_engine(3), region.id, b"k9", b"v1");
}