Method get_multi_with_consistency

d-engine-server/src/api/embedded_client.rs:362–403

Advanced: Get multiple keys with explicit consistency policy.

(
        &self,
        keys: &[Bytes],
        consistency: ReadConsistencyPolicy,
    )

    /// Advanced: Get multiple keys with explicit consistency policy.
    pub async fn get_multi_with_consistency(
        &self,
        keys: &[Bytes],
        consistency: ReadConsistencyPolicy,
    ) -> ClientApiResult<Vec<Option<Bytes>>> {
        let request = ClientReadRequest {
            client_id: self.client_id,
            keys: keys.to_vec(),
            consistency_policy: Some(consistency),
        };

        let (resp_tx, resp_rx) = MaybeCloneOneshot::new();

        self.cmd_tx
            .send(d_engine_core::ClientCmd::Read(request, resp_tx))
            .await
            .map_err(|_| channel_closed_error())?;

        let result = tokio::time::timeout(self.timeout, resp_rx)
            .await
            .map_err(|_| timeout_error(self.timeout))?
            .map_err(|_| channel_closed_error())?;

        let response =
            result.map_err(|status| server_error(format!("RPC error: {}", status.message())))?;

        if response.error != ErrorCode::Success {
            return Err(Self::map_error_response(
                response.error,
                response.leader_hint,
                response.retry_after_ms,
            ));
        }

        let read_results = extract_read_payload(response.result)?;
        // Reconstruct result vector in requested key order.
        // Server only returns results for keys that exist, so we must
        // map by key to preserve positional correspondence with input.
        let results_by_key: std::collections::HashMap<_, _> =
            read_results.entries.into_iter().map(|e| (e.key, e.value)).collect();
        Ok(keys.iter().map(|k| results_by_key.get(k).cloned()).collect())
    }

    /// Delete a key-value pair with strong consistency.
    ///