Replication timeout can duplicate a committed follower append on retry

[YZL0v3ZZ]

Describe the bug

At audited commit 82168d5d6fd6554a0b0be8589267ccbeea7a7986, ReplicationClient::submit() wraps both request handoff and ACK waiting in a single tokio::time::timeout. Once replication_request_tx.send(...).await succeeds, the request is already owned by the replication stream; a timeout only cancels the waiter. If the follower has already appended the batch and the ACK arrives late, the ACK is dropped by the canceled oneshot receiver. The leader then treats the replication error as recoverable, omits the subrequest from both successes and failures, and the router retries the same subrequest with a stale replica position. Because the follower-side stale-position branch is still a TODO, the same batch is appended again. On the retry, the leader appends locally and only afterward raises bad replica position, which is still treated as transient.

Relevant source anchors:

• ReplicationClient::submit:

  quickwit/quickwit/quickwit-ingest/src/ingest_v2/replication.rs

  Lines 370 to 391 in 82168d5

  	/// Submits a replication request to the replication stream and waits for the response.
  	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)?
  	}

• Leader replication error handling and omission from the response:

  quickwit/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs

  Lines 653 to 725 in 82168d5

  	// replicate to the follower
  	{
  	let mut replicate_futures = FuturesUnordered::new();
  	for (follower_id, replicate_subrequests) in per_follower_replicate_subrequests {
  	let replication_client = state_guard
  	.replication_streams
  	.get(&follower_id)
  	.expect("replication stream should be initialized")
  	.replication_client();
  	let leader_id = self.self_node_id.clone();
  	let replicate_future = replication_client.replicate(
  	leader_id,
  	follower_id,
  	replicate_subrequests,
  	commit_type,
  	);
  	replicate_futures.push(replicate_future);
  	}
  	while let Some(replication_result) = replicate_futures.next().await {
  	let replicate_response = match replication_result {
  	Ok(replicate_response) => replicate_response,
  	Err(_) => {
  	// TODO: Handle replication error:
  	// 1. Close and evict all the shards hosted by the follower.
  	// 2. Close and evict the replication client.
  	// 3. Return `PersistFailureReason::NodeUnavailable` to router.
  	continue;
  	}
  	};
  	for replicate_success in replicate_response.successes {
  	let pending_persist_subrequest = pending_persist_subrequests
  	.get_mut(&replicate_success.subrequest_id)
  	.expect("persist subrequest should exist");
  	pending_persist_subrequest.successfully_replicated = true;
  	pending_persist_subrequest.expected_position_inclusive =
  	replicate_success.replication_position_inclusive;
  	}
  	for replicate_failure in replicate_response.failures {
  	// TODO: If the replica shard is closed, close the primary shard if it is not
  	// already.
  	let persist_failure_reason: PersistFailureReason =
  	replicate_failure.reason().into();
  	let persist_failure = PersistFailure {
  	subrequest_id: replicate_failure.subrequest_id,
  	index_uid: replicate_failure.index_uid,
  	source_id: replicate_failure.source_id,
  	reason: persist_failure_reason as i32,
  	};
  	persist_failures.push(persist_failure);
  	}
  	}
  	}
  	// finally write locally
  	{
  	let now = Instant::now();
  	for subrequest in pending_persist_subrequests.into_values() {
  	if !subrequest.successfully_replicated {
  	continue;
  	}
  	let queue_id = subrequest.queue_id;
  	let batch_num_docs = subrequest.doc_batch.num_docs() as u64;
  	let append_result = append_non_empty_doc_batch(
  	&mut state_guard.mrecordlog,
  	&queue_id,
  	subrequest.doc_batch,
  	force_commit,
  	)
  	.await;

• Follower stale-position branch:

  quickwit/quickwit/quickwit-ingest/src/ingest_v2/replication.rs

  Lines 552 to 634 in 82168d5

  	for subrequest in replicate_request.subrequests {
  	let queue_id = subrequest.queue_id();
  	let from_position_exclusive = subrequest.from_position_exclusive();
  	let Some(shard) = state_guard.shards.get(&queue_id) else {
  	let replicate_failure = ReplicateFailure {
  	subrequest_id: subrequest.subrequest_id,
  	index_uid: subrequest.index_uid,
  	source_id: subrequest.source_id,
  	shard_id: subrequest.shard_id,
  	reason: ReplicateFailureReason::ShardNotFound as i32,
  	};
  	replicate_failures.push(replicate_failure);
  	continue;
  	};
  	assert!(shard.is_replica());
  	if shard.is_closed() {
  	let replicate_failure = ReplicateFailure {
  	subrequest_id: subrequest.subrequest_id,
  	index_uid: subrequest.index_uid,
  	source_id: subrequest.source_id,
  	shard_id: subrequest.shard_id,
  	reason: ReplicateFailureReason::ShardClosed as i32,
  	};
  	replicate_failures.push(replicate_failure);
  	continue;
  	}
  	if shard.replication_position_inclusive != from_position_exclusive {
  	// TODO
  	}
  	let doc_batch = match subrequest.doc_batch {
  	Some(doc_batch) if !doc_batch.is_empty() => doc_batch,
  	_ => {
  	warn!("received empty replicate request");
  	let replicate_success = ReplicateSuccess {
  	subrequest_id: subrequest.subrequest_id,
  	index_uid: subrequest.index_uid,
  	source_id: subrequest.source_id,
  	shard_id: subrequest.shard_id,
  	replication_position_inclusive: Some(
  	shard.replication_position_inclusive.clone(),
  	),
  	};
  	replicate_successes.push(replicate_success);
  	continue;
  	}
  	};
  	let batch_num_bytes = doc_batch.num_bytes() as u64;
  	let batch_num_docs = doc_batch.num_docs() as u64;
  	let requested_capacity = estimate_size(&doc_batch);
  	if let Err(error) = check_enough_capacity(
  	&state_guard.mrecordlog,
  	self.disk_capacity,
  	self.memory_capacity,
  	requested_capacity,
  	) {
  	rate_limited_warn!(
  	limit_per_min = 10,
  	"failed to replicate records to ingester `{}`: {error}",
  	self.follower_id,
  	);
  	let replicate_failure = ReplicateFailure {
  	subrequest_id: subrequest.subrequest_id,
  	index_uid: subrequest.index_uid,
  	source_id: subrequest.source_id,
  	shard_id: subrequest.shard_id,
  	reason: ReplicateFailureReason::WalFull as i32,
  	};
  	replicate_failures.push(replicate_failure);
  	continue;
  	};
  	let append_result = append_non_empty_doc_batch(
  	&mut state_guard.mrecordlog,
  	&queue_id,
  	doc_batch,
  	force_commit,
  	)
  	.await;

Steps to reproduce (if applicable)

1. Add the following whitebox regression test to quickwit-ingest/src/ingest_v2/ingester.rs.
2. Run:

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

3. On Linux x86, the command currently passes:

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

This passing result is intentional: the test asserts the current bad state, so a pass means the timeout/cancellation window is reachable.

Whitebox test

#[tokio::test]
async fn test_replicate_timeout_after_follower_append_retries_stale_position() {
    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_replicate_ack_held_tx, first_replicate_ack_held_rx) = oneshot::channel();
    let (release_first_replicate_ack_tx, release_first_replicate_ack_rx) = oneshot::channel();
    let (first_replicate_ack_forwarded_tx, first_replicate_ack_forwarded_rx) = oneshot::channel();
    let controlled_follower = HoldFirstReplicateAckIngester::new(
        follower.clone(),
        first_replicate_ack_held_tx,
        release_first_replicate_ack_rx,
        first_replicate_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 init_shards_request = InitShardsRequest {
        subrequests: vec![InitShardSubrequest {
            subrequest_id: 0,
            shard: Some(Shard {
                index_uid: Some(index_uid.clone()),
                source_id: source_id.clone(),
                shard_id: Some(shard_id.clone()),
                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()
            }),
            doc_mapping_json,
            validate_docs: true,
        }],
    };
    leader.init_shards(init_shards_request).await.unwrap();

    {
        let leader_state_guard = leader.state.lock_fully("test").await.unwrap();
        let primary_shard = leader_state_guard.shards.get(&queue_id).unwrap();
        primary_shard.assert_is_primary();
        primary_shard.assert_replication_position(Position::Beginning);
        leader_state_guard
            .mrecordlog
            .assert_records_eq(&queue_id, .., &[]);
    }
    {
        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_replication_position(Position::Beginning);
        follower_state_guard
            .mrecordlog
            .assert_records_eq(&queue_id, .., &[]);
    }

    let make_persist_request = || PersistRequest {
        leader_id: leader_ctx.node_id.to_string(),
        commit_type: CommitTypeV2::Auto as i32,
        subrequests: vec![PersistSubrequest {
            subrequest_id: 0,
            index_uid: Some(index_uid.clone()),
            source_id: source_id.clone(),
            doc_batch: Some(DocBatchV2::for_test([r#"{"doc": "detached-doc"}"#])),
        }],
    };

    let first_persist_handle = tokio::spawn({
        let leader = leader.clone();
        let persist_request = make_persist_request();
        async move { leader.persist(persist_request).await }
    });

    timeout(Duration::from_secs(2), first_replicate_ack_held_rx)
        .await
        .expect("first replicate ACK should be held")
        .expect("first replicate 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_replication_position(Position::offset(0u64));
        follower_state_guard.mrecordlog.assert_records_eq(
            &queue_id,
            ..,
            &[(0, [0, 0], r#"{"doc": "detached-doc"}"#)],
        );
    }

    let first_persist_response = timeout(Duration::from_secs(2), first_persist_handle)
        .await
        .expect("first persist should finish after replication timeout")
        .expect("first persist task should not panic")
        .expect("replication timeout is swallowed into an incomplete PersistResponse");
    assert_eq!(first_persist_response.successes.len(), 0);
    assert_eq!(first_persist_response.failures.len(), 0);

    {
        let leader_state_guard = leader.state.lock_fully("test").await.unwrap();
        let primary_shard = leader_state_guard.shards.get(&queue_id).unwrap();
        primary_shard.assert_replication_position(Position::Beginning);
        leader_state_guard
            .mrecordlog
            .assert_records_eq(&queue_id, .., &[]);
    }

    release_first_replicate_ack_tx
        .send(())
        .expect("first replicate ACK release should be received");
    timeout(Duration::from_secs(2), first_replicate_ack_forwarded_rx)
        .await
        .expect("late replicate ACK should be forwarded")
        .expect("late replicate ACK forwarded signal should be sent");

    let retry_error = leader
        .persist(make_persist_request())
        .await
        .expect_err("retry should fail after appending locally with a bad replica position");
    let IngestV2Error::Internal(error_message) = retry_error else {
        panic!("expected bad replica position internal error");
    };
    assert!(
        error_message.contains("bad replica position"),
        "unexpected internal error: {error_message}"
    );

    {
        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_replication_position(Position::offset(1u64));
        follower_state_guard.mrecordlog.assert_records_eq(
            &queue_id,
            ..,
            &[
                (0, [0, 0], r#"{"doc": "detached-doc"}"#),
                (1, [0, 0], r#"{"doc": "detached-doc"}"#),
            ],
        );
    }
    {
        let leader_state_guard = leader.state.lock_fully("test").await.unwrap();
        let primary_shard = leader_state_guard.shards.get(&queue_id).unwrap();
        primary_shard.assert_replication_position(Position::Beginning);
        leader_state_guard.mrecordlog.assert_records_eq(
            &queue_id,
            ..,
            &[(0, [0, 0], r#"{"doc": "detached-doc"}"#)],
        );
    }
}

What this test proves:

• the first request is held until the replication timeout fires;
• the follower append has already committed when the first ACK is finally released;
• the late ACK cannot revive the canceled waiter;
• the retry reuses the leader's stale replication position;
• the follower appends the same batch again;
• the leader appends locally, then fails with bad replica position.

Expected behavior

A replication timeout after request handoff should represent an unknown outcome, not a proof that nothing happened. Quickwit should either reconcile the late ACK/committed state, reject stale retries before any second append, or preserve the original outcome so the same batch cannot be committed again.

Configuration

1. quickwit --version: 0.8.0 (audited workspace version)
2. index_config.yaml: N/A for this unit-level whitebox reproducer

Audited source commit: 82168d5d6fd6554a0b0be8589267ccbeea7a7986