Replica init timeout can leave a leader WAL queue without a primary shard and poison retry

[YZL0v3ZZ]

Describe the bug

At audited commit 82168d5d6fd6554a0b0be8589267ccbeea7a7986, replicated shard initialization can leave the leader in a partially committed state if the follower replica init response is not observed before the replication timeout.

The problematic transition is:

leader starts replicated InitShard for queue Q
-> leader creates WAL queue Q
-> leader awaits follower init_replica through ReplicationClient::submit()
-> the wait times out or returns an error before the leader inserts the primary shard
-> init_primary_shard returns InitShardFailure without deleting Q
-> retry for the same shard sees state.shards[Q] vacant
-> retry calls create_queue(Q) again and fails with AlreadyExists

This is not a generic "follower init failed" case. The leader has already crossed a local WAL side-effect boundary, but the commit marker in state.shards is only written after the remote init succeeds.

Relevant source facts:

• init_primary_shard() creates the leader WAL queue before awaiting follower init, and the error branch explicitly lacks cleanup:
  quickwit-ingest/src/ingest_v2/ingester.rs#L169-L249

  let Entry::Vacant(entry) = state.shards.entry(queue_id.clone()) else {
      return Ok(());
  };
  // ...
  match mrecordlog.create_queue(&queue_id).await {
      Ok(_) => {}
      Err(CreateQueueError::AlreadyExists) => {
          let message = format!("WAL queue `{queue_id}` already exists");
          return Err(IngestV2Error::Internal(message));
      }
      // ...
  };
  // ...
  let primary_shard = if let Some(follower_id) = &shard.follower_id {
      // ...
      if let Err(error) = replication_client.init_replica(shard).await {
          // TODO: Remove dangling queue from the WAL.
          let message = format!("failed to initialize replica shard: {error}");
          return Err(IngestV2Error::Internal(message));
      }
      IngesterShard::new_primary(index_uid, source_id, shard_id, follower_id)
          // ...
          .build()
  } else {
      // ...
  };
  entry.insert(primary_shard);

• init_replica() is implemented through submit(), which wraps both request handoff and response waiting in tokio::time::timeout:
  quickwit-ingest/src/ingest_v2/replication.rs#L315-L391

  pub fn init_replica(self, replica_shard: Shard) -> impl Future<Output = Result<InitReplicaResponse, ReplicationError>> + Send + 'static {
      // ...
      async {
          self.submit(replication_request)
              .await
              .map(|replication_response| { /* Init response expected */ })
      }
  }
  
  fn submit(self, replication_request: ReplicationRequest) -> impl Future<Output = Result<ReplicationResponse, ReplicationError>> + Send + 'static {
      let (oneshot_replication_response_tx, oneshot_replication_response_rx) = oneshot::channel();
  
      let send_recv_fut = async move {
          self.replication_request_tx
              .send((replication_request, oneshot_replication_response_tx))
              .await
              .map_err(|_| ReplicationError::Closed)?;
          let replicate_response = oneshot_replication_response_rx
              .await
              .map_err(|_| ReplicationError::Closed)?;
          Ok(replicate_response)
      };
      async {
          tokio::time::timeout(REPLICATION_REQUEST_TIMEOUT, send_recv_fut)
              .await
              .map_err(|_| ReplicationError::Timeout)?
      }
  }

• If the init request reached the follower before the leader timed out, the follower may already have created its replica queue and inserted the replica shard:
  quickwit-ingest/src/ingest_v2/replication.rs#L431-L481

  match state_guard.mrecordlog.create_queue(&queue_id).await {
      Ok(_) => {}
      Err(CreateQueueError::AlreadyExists) => {
          let message = format!("WAL queue `{queue_id}` already exists");
          return Err(IngestV2Error::Internal(message));
      }
      // ...
  };
  // ...
  let replica_shard =
      IngesterShard::new_replica(index_uid, source_id, shard_id, leader_id).build();
  state_guard.shards.insert(queue_id, replica_shard);

Tokio documents that timeout returns an error and cancels the wrapped future when the duration elapses: https://docs.rs/tokio/1.52.3/tokio/time/fn.timeout.html. After the replication request has been handed to the background stream, timing out the local waiter does not roll back the already-created leader queue, nor does it undo follower work that may already have completed.

The demonstrated issue is initialization non-atomicity and retry poisoning for the same shard identity. I am not claiming document loss or duplicate indexed documents from this specific path.

Steps to reproduce (if applicable)

1. Add the following whitebox test code inside #[cfg(test)] mod tests in quickwit-ingest/src/ingest_v2/ingester.rs. The existing test module imports and use super::* are sufficient for the referenced symbols.
2. Run the targeted test command shown below.
3. Observe that the bug-existence test passes, meaning the current bad state was reached and asserted.

The test drives the production path as follows:

1. Start a leader and follower ingester with replication enabled.
2. Wrap the follower with a test IngesterService that forwards the first InitResponse only after a gate is released.
3. Start a replicated InitShardsRequest.
4. Wait until the follower has already initialized the replica queue and shard.
5. Let the leader-side init_replica() waiter time out.
6. Assert that the first init returns one InitShardFailure.
7. Assert that the leader WAL queue exists while the leader has no primary shard entry for the same queue.
8. Retry the same shard through init_primary_shard() and assert that it fails with already exists.

Whitebox test code for quickwit-ingest/src/ingest_v2/ingester.rs

#[derive(Clone)]
struct HoldFirstInitAckIngester {
    inner: Ingester,
    first_init_ack_held_tx: Arc<std::sync::Mutex<Option<oneshot::Sender<()>>>>,
    release_first_init_ack_rx: Arc<std::sync::Mutex<Option<oneshot::Receiver<()>>>>,
    first_init_ack_forwarded_tx: Arc<std::sync::Mutex<Option<oneshot::Sender<()>>>>,
}

impl fmt::Debug for HoldFirstInitAckIngester {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("HoldFirstInitAckIngester")
            .finish_non_exhaustive()
    }
}

impl HoldFirstInitAckIngester {
    fn new(
        inner: Ingester,
        first_init_ack_held_tx: oneshot::Sender<()>,
        release_first_init_ack_rx: oneshot::Receiver<()>,
        first_init_ack_forwarded_tx: oneshot::Sender<()>,
    ) -> Self {
        Self {
            inner,
            first_init_ack_held_tx: Arc::new(std::sync::Mutex::new(Some(
                first_init_ack_held_tx,
            ))),
            release_first_init_ack_rx: Arc::new(std::sync::Mutex::new(Some(
                release_first_init_ack_rx,
            ))),
            first_init_ack_forwarded_tx: Arc::new(std::sync::Mutex::new(Some(
                first_init_ack_forwarded_tx,
            ))),
        }
    }
}

#[async_trait::async_trait]
impl IngesterService for HoldFirstInitAckIngester {
    async fn persist(&self, request: PersistRequest) -> IngestV2Result<PersistResponse> {
        self.inner.persist(request).await
    }

    async fn open_replication_stream(
        &self,
        request: ServiceStream<SynReplicationMessage>,
    ) -> IngestV2Result<IngesterServiceStream<AckReplicationMessage>> {
        let mut inner_ack_stream = self.inner.open_replication_stream(request).await?;
        let (controlled_ack_tx, controlled_ack_stream) = ServiceStream::new_unbounded();

        let mut first_init_ack_held_tx = self.first_init_ack_held_tx.lock().unwrap().take();
        let mut release_first_init_ack_rx =
            self.release_first_init_ack_rx.lock().unwrap().take();
        let mut first_init_ack_forwarded_tx =
            self.first_init_ack_forwarded_tx.lock().unwrap().take();

        tokio::spawn(async move {
            while let Some(ack_message_result) = inner_ack_stream.next().await {
                let should_hold_first_init_ack = first_init_ack_held_tx.is_some()
                    && matches!(
                        &ack_message_result,
                        Ok(AckReplicationMessage {
                            message: Some(ack_replication_message::Message::InitResponse(_)),
                            ..
                        })
                    );

                if should_hold_first_init_ack {
                    if let Some(held_tx) = first_init_ack_held_tx.take() {
                        let _ = held_tx.send(());
                    }
                    if let Some(release_rx) = release_first_init_ack_rx.take() {
                        let _ = release_rx.await;
                    }
                }

                if controlled_ack_tx.send(ack_message_result).is_err() {
                    return;
                }

                if should_hold_first_init_ack
                    && let Some(forwarded_tx) = first_init_ack_forwarded_tx.take()
                {
                    let _ = forwarded_tx.send(());
                }
            }
        });

        Ok(controlled_ack_stream)
    }

    async fn open_fetch_stream(
        &self,
        request: OpenFetchStreamRequest,
    ) -> IngestV2Result<IngesterServiceStream<FetchMessage>> {
        self.inner.open_fetch_stream(request).await
    }

    async fn open_observation_stream(
        &self,
        request: OpenObservationStreamRequest,
    ) -> IngestV2Result<IngesterServiceStream<ObservationMessage>> {
        self.inner.open_observation_stream(request).await
    }

    async fn init_shards(
        &self,
        request: InitShardsRequest,
    ) -> IngestV2Result<InitShardsResponse> {
        self.inner.init_shards(request).await
    }

    async fn retain_shards(
        &self,
        request: RetainShardsRequest,
    ) -> IngestV2Result<RetainShardsResponse> {
        self.inner.retain_shards(request).await
    }

    async fn truncate_shards(
        &self,
        request: TruncateShardsRequest,
    ) -> IngestV2Result<TruncateShardsResponse> {
        self.inner.truncate_shards(request).await
    }

    async fn close_shards(
        &self,
        request: CloseShardsRequest,
    ) -> IngestV2Result<CloseShardsResponse> {
        self.inner.close_shards(request).await
    }

    async fn decommission(
        &self,
        request: DecommissionRequest,
    ) -> IngestV2Result<DecommissionResponse> {
        self.inner.decommission(request).await
    }
}

#[tokio::test]
async fn test_init_replica_timeout_leaves_leader_queue_without_primary_shard() {
    let (leader_ctx, leader) = IngesterForTest::default()
        .with_node_id("test-leader")
        .with_replication()
        .build()
        .await;

    let (follower_ctx, follower) = IngesterForTest::default()
        .with_node_id("test-follower")
        .with_ingester_pool(&leader_ctx.ingester_pool)
        .with_replication()
        .build()
        .await;

    let (first_init_ack_held_tx, first_init_ack_held_rx) = oneshot::channel();
    let (release_first_init_ack_tx, release_first_init_ack_rx) = oneshot::channel();
    let (first_init_ack_forwarded_tx, first_init_ack_forwarded_rx) = oneshot::channel();
    let controlled_follower = HoldFirstInitAckIngester::new(
        follower.clone(),
        first_init_ack_held_tx,
        release_first_init_ack_rx,
        first_init_ack_forwarded_tx,
    );

    let ingester_pool_entry = IngesterPoolEntry {
        client: IngesterServiceClient::new(controlled_follower),
        status: IngesterStatus::Ready,
        availability_zone: None,
    };
    leader_ctx
        .ingester_pool
        .insert(follower_ctx.node_id.clone(), ingester_pool_entry);

    let index_uid = IndexUid::for_test("test-index", 0);
    let source_id = SourceId::from("test-source");
    let shard_id = ShardId::from(1);
    let queue_id = queue_id(&index_uid, &source_id, &shard_id);

    let doc_mapping_uid = DocMappingUid::random();
    let doc_mapping_json = format!(
        r#"{{
            "doc_mapping_uid": "{doc_mapping_uid}"
        }}"#
    );
    let shard = Shard {
        index_uid: Some(index_uid.clone()),
        source_id: source_id.clone(),
        shard_id: Some(shard_id),
        shard_state: ShardState::Open as i32,
        leader_id: leader_ctx.node_id.to_string(),
        follower_id: Some(follower_ctx.node_id.to_string()),
        doc_mapping_uid: Some(doc_mapping_uid),
        ..Default::default()
    };
    let init_shards_request = InitShardsRequest {
        subrequests: vec![InitShardSubrequest {
            subrequest_id: 0,
            shard: Some(shard.clone()),
            doc_mapping_json: doc_mapping_json.clone(),
            validate_docs: true,
        }],
    };

    let first_init_handle = tokio::spawn({
        let leader = leader.clone();
        async move { leader.init_shards(init_shards_request).await }
    });

    timeout(Duration::from_secs(2), first_init_ack_held_rx)
        .await
        .expect("first init ACK should be held")
        .expect("first init ACK held signal should be sent");

    {
        let follower_state_guard = follower.state.lock_fully("test").await.unwrap();
        let replica_shard = follower_state_guard.shards.get(&queue_id).unwrap();
        replica_shard.assert_is_replica();
        replica_shard.assert_is_open();
        replica_shard.assert_replication_position(Position::Beginning);
        assert!(follower_state_guard.mrecordlog.queue_exists(&queue_id));
    }

    let first_init_response = timeout(Duration::from_secs(2), first_init_handle)
        .await
        .expect("leader init should finish after replica init timeout")
        .expect("leader init task should not panic")
        .expect("replica init timeout is reported as an InitShard failure");
    assert_eq!(first_init_response.successes.len(), 0);
    assert_eq!(first_init_response.failures.len(), 1);
    assert_eq!(first_init_response.failures[0].subrequest_id, 0);

    {
        let leader_state_guard = leader.state.lock_fully("test").await.unwrap();
        assert!(leader_state_guard.mrecordlog.queue_exists(&queue_id));
        assert!(
            !leader_state_guard.shards.contains_key(&queue_id),
            "the leader WAL queue exists but the primary shard was never committed"
        );
    }

    let retry_error = {
        let mut leader_state_guard = leader.state.lock_fully("test").await.unwrap();
        leader
            .init_primary_shard(
                &mut leader_state_guard.inner,
                &mut leader_state_guard.mrecordlog,
                shard,
                &doc_mapping_json,
                Instant::now(),
                true,
            )
            .await
            .expect_err("retrying the same shard should hit the dangling WAL queue")
    };
    let IngestV2Error::Internal(error_message) = retry_error else {
        panic!("expected an internal AlreadyExists error on same-shard retry");
    };
    assert!(
        error_message.contains("already exists"),
        "unexpected retry error: {error_message}"
    );

    {
        let leader_state_guard = leader.state.lock_fully("test").await.unwrap();
        assert!(leader_state_guard.mrecordlog.queue_exists(&queue_id));
        assert!(!leader_state_guard.shards.contains_key(&queue_id));
    }

    release_first_init_ack_tx
        .send(())
        .expect("first init ACK release should be received");
    timeout(Duration::from_secs(2), first_init_ack_forwarded_rx)
        .await
        .expect("late init ACK should be forwarded")
        .expect("late init ACK forwarded signal should be sent");
}

Command:

cargo test -p quickwit-ingest --lib test_init_replica_timeout_leaves_leader_queue_without_primary_shard -- --nocapture

Observed result:

jcj@server36:~/whitebox/quickwit/quickwit$ cargo test -p quickwit-ingest --lib test_init_replica_timeout_leaves_leader_queue_without_primary_shard -- --nocapture
   Compiling quickwit-ingest v0.8.0 (/home/jcj/whitebox/quickwit/quickwit/quickwit-ingest)
    Finished `test` profile [unoptimized] target(s) in 10.58s
     Running unittests src/lib.rs (target/debug/deps/quickwit_ingest-b4676b75865f9fb5)

running 1 test
test ingest_v2::ingester::tests::test_init_replica_timeout_leaves_leader_queue_without_primary_shard ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 173 filtered out; finished in 0.27s

jcj@server36:~/whitebox/quickwit/quickwit$

This is a bug-existence test. A passing result means the current implementation reached the asserted bad state:

• follower replica queue and shard were initialized;
• leader returned an InitShardFailure;
• leader WAL queue existed without a leader primary shard entry;
• same-shard retry failed with already exists.

The reproducer is unit-level and controlled: it uses local test ingesters, temporary WAL state, and a deterministic in-process ACK gate. It does not require external services, credentials, or production data.

Expected behavior

Replicated shard initialization should not expose a failed init result while leaving a leader WAL queue that cannot be retried or adopted.

After the leader creates WAL queue Q, every timeout/error before successful primary shard publication should guarantee at least one of these invariants:

1. Q is rolled back before returning the init failure.
2. A primary shard entry for Q is committed or an explicit initializing/unknown state is recorded.
3. Retry of the same shard identity reconciles or adopts the existing Q instead of treating AlreadyExists as fatal.
4. A late follower init success is reconciled before the leader retries or reports a terminal failure.

Reasonable repair directions include implementing the existing cleanup TODO with careful error handling, introducing an initializing shard state, making same-identity queue creation/adoption idempotent, or treating post-handoff replica-init timeout as outcome-unknown and reconciling local/follower state before retry.

Configuration

Please provide:

1. Output of quickwit --version
2. The index_config.yaml

For this report:

1. quickwit --version: source workspace version 0.8.0; audited source commit 82168d5d6fd6554a0b0be8589267ccbeea7a7986.
2. index_config.yaml: N/A. This is a unit-level whitebox reproducer for quickwit-ingest.