Repository: qdrant/qdrant
File: lib/collection/src/shards/forward_proxy_shard.rs (Rust)

Code before:
                    .wrapped_shard
                    .scroll_by(
                        None,
                        point_ids.len(),
                        &WithPayloadInterface::Bool(false),
                        &WithVector::Bool(false),
                        Some(&filter.with_point_ids(point_ids)),
                        &Handle::current(),
                        None,
                        None,                           // No timeout
                        HwMeasurementAcc::disposable(), // Internal operation, no need to measure hardware here?
                    )
                    .await?
                    .into_iter()
                    .map(|record| record.id)
                    .collect::<HashSet<_>>();

                let mut modified_operation = operation.clone();
                modified_operation.operation.retain_point_ids(|point_id| {
                    points_matching_filter_before.contains(point_id)
                        || points_matching_filter_after.contains(point_id)
                });

                Some(modified_operation)
            } else {
                Some(operation) // No point IDs to filter, forward as-is
            }
        } else {
            // If `ForwardProxyShard` `resharding_hash_ring` and `filter` are `None`, we assume that proxy is used
            // during *regular* shard transfer, so operation can be forwarded as-is, without any
            // additional handling.

            debug_assert_eq!(self.shard_id, self.remote_shard.id);

            Some(operation)
        };

        if let Some(operation) = forward_operation {
            let remote_result = self
                .remote_shard
                .update(operation, false, hw_measurement_acc)
                .await
                .map_err(|err| {
                    CollectionError::forward_proxy_error(self.remote_shard.peer_id, err)
                })?;

            // Merge `result` and `remote_result`:
            //
            // - Pick `clock_tag` with *newer* `clock_tick`
            let tick = result.clock_tag.map(|tag| tag.clock_tick);
            let remote_tick = remote_result.clock_tag.map(|tag| tag.clock_tick);

Review comment (suggestion):
If we send operations to other shard IDs, we must remove the clock tag in both cases.

I'd recommend to do it in the same way as we do for resharding. We can probably do it in a central place after `let forward_operation`.

If I recommended something else during the daily - I might have misunderstood.

Code after:
        let remote_result = self
            .remote_shard
            .update(operation, false, hw_measurement_acc)
            .await
            .map_err(|err| CollectionError::forward_proxy_error(self.remote_shard.peer_id, err))?;

        // Merge `result` and `remote_result`:
        //
        // - Pick `clock_tag` with *newer* `clock_tick`
        let tick = result.clock_tag.map(|tag| tag.clock_tick);
        let remote_tick = remote_result.clock_tag.map(|tag| tag.clock_tick);

        if remote_tick > tick || tick.is_none() {
            result.clock_tag = remote_result.clock_tag;
        }

        // - If any node *rejected* the operation, propagate `UpdateStatus::ClockRejected`
        if remote_result.status == UpdateStatus::ClockRejected {
            result.status = UpdateStatus::ClockRejected;
        }

        Ok(result)
    }

    /// Forward read-only `scroll_by` to `wrapped_shard`
    async fn scroll_by(
        &self,
        request: Arc<ScrollRequestInternal>,
        search_runtime_handle: &Handle,
        timeout: Option<Duration>,
        hw_measurement_acc: HwMeasurementAcc,
    ) -> CollectionResult<Vec<RecordInternal>> {
        let local_shard = &self.wrapped_shard;
        local_shard
            .scroll_by(request, search_runtime_handle, timeout, hw_measurement_acc)
            .await
    }

    async fn local_scroll_by_id(
        &self,
        offset: Option<ExtendedPointId>,
        limit: usize,
        with_payload_interface: &WithPayloadInterface,
        with_vector: &WithVector,
        filter: Option<&Filter>,
        search_runtime_handle: &Handle,
        timeout: Option<Duration>,
        hw_measurement_acc: HwMeasurementAcc,
    ) -> CollectionResult<Vec<RecordInternal>> {
        let local_shard = &self.wrapped_shard;
        local_shard