[bug] control plane: create_index never replies while a Ready indexer's ApplyIndexingPlan RPC is pending #6841

//! Reproduction for: the control plane never replies to `create_index` while a Ready
//! indexer leaves its `ApplyIndexingPlan` RPC pending.
//!
//! Paste into `quickwit/quickwit-control-plane/tests/control_plane_create_index_rebuild_hang.rs`.

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

use async_trait::async_trait;
use quickwit_actors::{ActorHandle, Mailbox, Supervisor, Universe};
use quickwit_config::{ClusterConfig, IndexConfig, KafkaSourceParams, SourceConfig, SourceParams};
use quickwit_control_plane::control_plane::ControlPlane;
use quickwit_control_plane::{IndexerPool, IndexerPoolEntry};
use quickwit_ingest::IngesterPool;
use quickwit_metastore::{CreateIndexRequestExt, IndexMetadata, ListIndexesMetadataResponseExt};
use quickwit_proto::indexing::{
    ApplyIndexingPlanRequest, ApplyIndexingPlanResponse, CpuCapacity, IndexingResult,
    IndexingService, IndexingServiceClient,
};
use quickwit_proto::ingest::ingester::IngesterStatus;
use quickwit_proto::metastore::{
    CreateIndexRequest, CreateIndexResponse, ListIndexesMetadataRequest,
    ListIndexesMetadataResponse, MetastoreServiceClient, MockMetastoreService,
};
use quickwit_proto::types::NodeId;

/// An `IndexingService` whose `apply_indexing_plan` future never resolves.
#[derive(Debug)]
struct HangingIndexingService {
    apply_count: Arc<AtomicUsize>,
}

#[async_trait]
impl IndexingService for HangingIndexingService {
    async fn apply_indexing_plan(
        &self,
        _request: ApplyIndexingPlanRequest,
    ) -> IndexingResult<ApplyIndexingPlanResponse> {
        self.apply_count.fetch_add(1, Ordering::SeqCst);
        std::future::pending().await
    }
}

/// An `IndexingService` that acknowledges plans immediately.
#[derive(Debug)]
struct RespondingIndexingService {
    apply_count: Arc<AtomicUsize>,
}

#[async_trait]
impl IndexingService for RespondingIndexingService {
    async fn apply_indexing_plan(
        &self,
        _request: ApplyIndexingPlanRequest,
    ) -> IndexingResult<ApplyIndexingPlanResponse> {
        self.apply_count.fetch_add(1, Ordering::SeqCst);
        Ok(ApplyIndexingPlanResponse {})
    }
}

fn kafka_source_config() -> SourceConfig {
    SourceConfig::for_test(
        "test-source",
        SourceParams::Kafka(KafkaSourceParams {
            topic: "test-topic".to_string(),
            client_log_level: None,
            client_params: serde_json::json!({
                "bootstrap.servers": "localhost:9092",
            }),
            enable_backfill_mode: false,
        }),
    )
}

/// Spawns a control plane with one Ready indexer and a metastore that creates an index carrying
/// one schedulable Kafka source, then waits for the control plane to become ready.
async fn spawn_control_plane(
    universe: &Universe,
    indexer_client: IndexingServiceClient,
) -> (
    Mailbox<ControlPlane>,
    CreateIndexRequest,
    ActorHandle<Supervisor<ControlPlane>>,
) {
    let indexer_node_id = NodeId::from_str("test-indexer");
    let indexer_pool = IndexerPool::default();
    indexer_pool.insert(
        indexer_node_id.clone(),
        IndexerPoolEntry {
            node_id: indexer_node_id,
            generation_id: 0,
            client: indexer_client,
            indexing_tasks: Vec::new(),
            indexing_capacity: CpuCapacity::from_cpu_millis(4_000),
            ingester_status: IngesterStatus::Ready,
            availability_zone: None,
        },
    );

    let source_config = kafka_source_config();
    let mut index_metadata = IndexMetadata::for_test("test-index", "ram:///indexes/test-index");
    index_metadata.add_source(source_config.clone()).unwrap();
    let index_uid = index_metadata.index_uid.clone();
    let index_metadata_json = serde_json::to_string(&index_metadata).unwrap();

    let mut mock_metastore = MockMetastoreService::new();
    mock_metastore
        .expect_list_indexes_metadata()
        .returning(|request: ListIndexesMetadataRequest| {
            assert_eq!(request, ListIndexesMetadataRequest::all());
            Ok(ListIndexesMetadataResponse::for_test(Vec::new()))
        });
    mock_metastore.expect_create_index().returning(move |_request| {
        Ok(CreateIndexResponse {
            index_uid: Some(index_uid.clone()),
            index_metadata_json: index_metadata_json.clone(),
        })
    });

    let index_config = IndexConfig::for_test("test-index", "ram:///indexes/test-index");
    let create_index_request =
        CreateIndexRequest::try_from_index_and_source_configs(&index_config, &[source_config])
            .unwrap();

    let (control_plane_mailbox, control_plane_handle, mut readiness_rx) = ControlPlane::spawn(
        universe,
        ClusterConfig::for_test(),
        NodeId::from_str("test-control-plane"),
        indexer_pool,
        IngesterPool::default(),
        MetastoreServiceClient::from_mock(mock_metastore),
    );
    tokio::time::timeout(
        Duration::from_secs(5),
        readiness_rx.wait_for(|readiness| *readiness),
    )
    .await
    .expect("control plane should become ready")
    .expect("readiness watch should stay open");
    (
        control_plane_mailbox,
        create_index_request,
        control_plane_handle,
    )
}

/// Demonstrates that one unresponsive Ready indexer is enough to keep the `create_index` reply
/// pending forever.
#[tokio::test]
async fn create_index_reply_is_blocked_by_a_hanging_ready_indexer_apply() {
    let universe = Universe::with_accelerated_time();
    let apply_count = Arc::new(AtomicUsize::new(0));
    let indexer_client = IndexingServiceClient::tower().build(HangingIndexingService {
        apply_count: apply_count.clone(),
    });
    let (control_plane_mailbox, create_index_request, _control_plane_handle) =
        spawn_control_plane(&universe, indexer_client).await;

    let create_index_task =
        tokio::spawn(async move { control_plane_mailbox.ask(create_index_request).await });

    // Advance the accelerated clock until the plan rebuild dispatches `apply_indexing_plan` to
    // the Ready indexer.
    let mut apply_dispatched = false;
    for _ in 0..80 {
        if apply_count.load(Ordering::SeqCst) > 0 {
            apply_dispatched = true;
            break;
        }
        universe.sleep(Duration::from_millis(250)).await;
    }
    assert!(
        apply_dispatched,
        "the control plane should dispatch `apply_indexing_plan` to the Ready indexer"
    );
    println!(
        "[apply] dispatched {} `apply_indexing_plan` call(s) to the hanging Ready indexer",
        apply_count.load(Ordering::SeqCst)
    );

    // The apply future never completes, so the rebuild generation is never published and the
    // deferred `create_index` reply must remain pending.
    for _ in 0..80 {
        if create_index_task.is_finished() {
            break;
        }
        universe.sleep(Duration::from_millis(250)).await;
    }
    assert!(
        !create_index_task.is_finished(),
        "the `create_index` reply resolved even though a Ready indexer apply never completed"
    );
    println!(
        "[blocked] `create_index` reply is still pending after {} apply call(s); the index \
         itself was already created in the metastore",
        apply_count.load(Ordering::SeqCst)
    );
    universe.assert_quit().await;
}

/// Control case: with a responsive indexer, the `create_index` reply does resolve.
#[tokio::test]
async fn create_index_reply_resolves_when_the_ready_indexer_answers() {
    let universe = Universe::with_accelerated_time();
    let apply_count = Arc::new(AtomicUsize::new(0));
    let indexer_client = IndexingServiceClient::tower().build(RespondingIndexingService {
        apply_count: apply_count.clone(),
    });
    let (control_plane_mailbox, create_index_request, _control_plane_handle) =
        spawn_control_plane(&universe, indexer_client).await;

    let create_index_task =
        tokio::spawn(async move { control_plane_mailbox.ask(create_index_request).await });

    let mut resolved = false;
    for _ in 0..80 {
        if create_index_task.is_finished() {
            resolved = true;
            break;
        }
        universe.sleep(Duration::from_millis(250)).await;
    }
    assert!(
        resolved,
        "with a responsive indexer the `create_index` reply must resolve"
    );
    let reply = create_index_task.await.unwrap();
    assert!(reply.is_ok(), "the reply channel should stay open");
    assert!(reply.unwrap().is_ok(), "the created index should be reported");
    println!(
        "[control] `create_index` reply resolved after {} apply call(s)",
        apply_count.load(Ordering::SeqCst)
    );
    universe.assert_quit().await;
}