Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion d-engine-client/src/error_test.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use d_engine_proto::error::ErrorCode;
use d_engine_core::client::ErrorCode;
use tonic::Code;
use tonic::Status;

Expand Down
14 changes: 8 additions & 6 deletions d-engine-client/src/grpc_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,18 +3,18 @@ use std::sync::Arc;
use arc_swap::ArcSwap;
use bytes::Bytes;
use d_engine_core::ScanResult;
use d_engine_core::client::ErrorCode;
use d_engine_core::client::KvEntry;
use d_engine_core::config::ReadConsistencyPolicy;
use d_engine_proto::client::ClientReadRequest;
use d_engine_proto::client::ClientResult;
use d_engine_proto::client::ClientWriteRequest;
use d_engine_proto::client::MembershipSnapshot;
use d_engine_proto::client::ReadConsistencyPolicy;
use d_engine_proto::client::ScanRequest;
use d_engine_proto::client::WatchMembershipRequest;
use d_engine_proto::client::WatchRequest;
use d_engine_proto::client::WatchResponse;
use d_engine_proto::client::WriteCommand;
use d_engine_proto::client::raft_client_service_client::RaftClientServiceClient;
use d_engine_proto::error::ErrorCode;
use rand::Rng;
use rand::SeedableRng;
use rand::rngs::StdRng;
Expand Down Expand Up @@ -56,7 +56,7 @@ impl GrpcClient {
&self,
key: impl AsRef<[u8]>,
consistency_policy: Option<ReadConsistencyPolicy>,
) -> std::result::Result<Option<ClientResult>, ClientApiError> {
) -> std::result::Result<Option<KvEntry>, ClientApiError> {
// Delegate to multi-get implementation
let mut results =
self.get_multi_with_policy(std::iter::once(key), consistency_policy).await?;
Expand All @@ -73,7 +73,7 @@ impl GrpcClient {
&self,
keys: impl IntoIterator<Item = impl AsRef<[u8]>>,
consistency_policy: Option<ReadConsistencyPolicy>,
) -> std::result::Result<Vec<Option<ClientResult>>, ClientApiError> {
) -> std::result::Result<Vec<Option<KvEntry>>, ClientApiError> {
let _timer = ScopedTimer::new("client::get_multi");

let client_inner = self.client_inner.load();
Expand All @@ -92,7 +92,9 @@ impl GrpcClient {
let request = ClientReadRequest {
client_id: client_inner.client_id,
keys,
consistency_policy: consistency_policy.map(|p| p as i32),
consistency_policy: consistency_policy
.clone()
.map(|p| d_engine_proto::client::ReadConsistencyPolicy::from(p) as i32),
};

// Select client based on policy (if specified)
Expand Down
130 changes: 109 additions & 21 deletions d-engine-client/src/grpc_client_test.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,12 @@ use arc_swap::ArcSwap;
use bytes::Bytes;
use d_engine_core::ClientApi;
use d_engine_core::client::ClientApiResult;
use d_engine_proto::client::ClientResponse;
use d_engine_proto::client::ClientResult;
use d_engine_proto::client::ReadConsistencyPolicy;
use d_engine_core::client::ClientResponse;
use d_engine_core::client::ErrorCode;
use d_engine_core::client::KvEntry;
use d_engine_core::config::ReadConsistencyPolicy;
use d_engine_proto::client::WatchEventType;
use d_engine_proto::client::WatchResponse;
use d_engine_proto::error::ErrorCode;
use d_engine_proto::server::cluster::ClusterMembership;
use tokio::sync::oneshot;
use tracing_test::traced_test;
Expand Down Expand Up @@ -219,7 +219,7 @@ async fn test_get_success() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: key.clone(),
value: value.clone(),
}]),
Expand Down Expand Up @@ -369,7 +369,7 @@ async fn test_get_multi_success_linear() {
// Create a single response containing all key-value pairs
let mut client_results = Vec::new();
for (i, key) in keys.iter().enumerate() {
client_results.push(ClientResult {
client_results.push(KvEntry {
key: key.clone(),
value: values[i].clone(),
});
Expand Down Expand Up @@ -438,7 +438,7 @@ async fn test_get_multi_success_non_linear() {
// Create a single response containing all key-value pairs
let mut client_results = Vec::new();
for (i, key) in keys.iter().enumerate() {
client_results.push(ClientResult {
client_results.push(KvEntry {
key: key.clone(),
value: match &values[i] {
Some(value) => value.clone(),
Expand Down Expand Up @@ -580,7 +580,7 @@ async fn test_get_linearizable_success() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: key.clone(),
value: value.clone(),
}]),
Expand Down Expand Up @@ -618,7 +618,7 @@ async fn test_get_lease_success() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: key.clone(),
value: value.clone(),
}]),
Expand Down Expand Up @@ -656,7 +656,7 @@ async fn test_get_eventual_success() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: key.clone(),
value: value.clone(),
}]),
Expand Down Expand Up @@ -699,7 +699,7 @@ async fn test_get_multi_success() {

let mut client_results = Vec::new();
for (i, key) in keys.iter().enumerate() {
client_results.push(ClientResult {
client_results.push(KvEntry {
key: key.clone(),
value: values[i].clone(),
});
Expand Down Expand Up @@ -752,11 +752,11 @@ async fn test_get_multi_with_mixed_results() {

// Only key1 and key3 exist, key2 is missing
let client_results = vec![
ClientResult {
KvEntry {
key: keys[0].clone(),
value: Bytes::from("value1".to_string()),
},
ClientResult {
KvEntry {
key: keys[2].clone(),
value: Bytes::from("value3".to_string()),
},
Expand Down Expand Up @@ -882,7 +882,7 @@ async fn test_client_refresh_with_new_endpoints() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key1"),
value: Bytes::from("value1"),
}]),
Expand Down Expand Up @@ -920,7 +920,7 @@ async fn test_client_refresh_with_new_endpoints() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key2"),
value: Bytes::from("value2"),
}]),
Expand Down Expand Up @@ -951,7 +951,7 @@ async fn test_client_refresh_with_none_endpoints() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key1"),
value: Bytes::from("value1"),
}]),
Expand Down Expand Up @@ -1003,7 +1003,7 @@ async fn test_client_refresh_with_multiple_endpoints() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key1"),
value: Bytes::from("value1"),
}]),
Expand All @@ -1017,7 +1017,7 @@ async fn test_client_refresh_with_multiple_endpoints() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key2"),
value: Bytes::from("value2"),
}]),
Expand Down Expand Up @@ -1077,7 +1077,7 @@ async fn test_client_refresh_failure_invalid_endpoints() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key1"),
value: Bytes::from("value1"),
}]),
Expand Down Expand Up @@ -1134,7 +1134,7 @@ async fn test_client_refresh_preserves_kv_and_cluster_clients() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key1"),
value: Bytes::from("value1"),
}]),
Expand Down Expand Up @@ -1168,7 +1168,7 @@ async fn test_client_refresh_preserves_kv_and_cluster_clients() {
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![ClientResult {
ClientResponse::read_results(vec![KvEntry {
key: Bytes::from("key2"),
value: Bytes::from("value2"),
}]),
Expand Down Expand Up @@ -1738,3 +1738,91 @@ mod scan_tests {
assert!(result.entries.is_empty());
}
}

// ─── Type boundary tests: get_with_policy / get_multi_with_policy must return KvEntry ──────

#[tokio::test]
#[traced_test]
async fn test_get_with_policy_returns_native_kv_entry() {
let key = Bytes::from("kv_key");
let value = Bytes::from("kv_value");
let (_tx, rx) = oneshot::channel::<()>();
let (_channel, port) = MockNode::simulate_client_read_mock_server(
rx,
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![KvEntry {
key: key.clone(),
value: value.clone(),
}]),
)
.await
.unwrap();

let endpoints = vec![format!("http://localhost:{}", port)];
let config = ClientConfig::default();
let pool = ConnectionPool::create(endpoints.clone(), config.clone()).await.unwrap();
let client = GrpcClient::new(Arc::new(ArcSwap::from_pointee(ClientInner {
pool,
client_id: 1,
config,
endpoints,
})));

// Explicit type annotation: return type must be KvEntry, not proto::ClientResult
let entry: Option<KvEntry> = client
.get_with_policy(key.clone(), Some(ReadConsistencyPolicy::LinearizableRead))
.await
.unwrap();

assert_eq!(entry, Some(KvEntry { key, value }));
}

#[tokio::test]
#[traced_test]
async fn test_get_multi_with_policy_returns_native_kv_entries() {
let k1 = Bytes::from("mk1");
let v1 = Bytes::from("mv1");
let k2 = Bytes::from("mk2");
let v2 = Bytes::from("mv2");
let (_tx, rx) = oneshot::channel::<()>();
let (_channel, port) = MockNode::simulate_client_read_mock_server(
rx,
None::<
Box<dyn Fn(u16) -> std::result::Result<ClusterMembership, tonic::Status> + Send + Sync>,
>,
ClientResponse::read_results(vec![
KvEntry {
key: k1.clone(),
value: v1.clone(),
},
KvEntry {
key: k2.clone(),
value: v2.clone(),
},
]),
)
.await
.unwrap();

let endpoints = vec![format!("http://localhost:{}", port)];
let config = ClientConfig::default();
let pool = ConnectionPool::create(endpoints.clone(), config.clone()).await.unwrap();
let client = GrpcClient::new(Arc::new(ArcSwap::from_pointee(ClientInner {
pool,
client_id: 1,
config,
endpoints,
})));

// Explicit type annotation: must be Vec<Option<KvEntry>>, not Vec<Option<ClientResult>>
let results: Vec<Option<KvEntry>> = client
.get_multi_with_policy(vec![k1.clone(), k2.clone()].into_iter(), None)
.await
.unwrap();

assert_eq!(results.len(), 2);
assert_eq!(results[0], Some(KvEntry { key: k1, value: v1 }));
assert_eq!(results[1], Some(KvEntry { key: k2, value: v2 }));
}
8 changes: 4 additions & 4 deletions d-engine-client/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,8 @@ mod utils;

pub use builder::*;
pub use config::*;
pub use d_engine_core::client::ErrorCode;
pub use d_engine_core::client::{ClientApi, ClientApiError, ClientApiResult};
pub use d_engine_proto::error::ErrorCode;
pub use grpc_client::*;
pub(crate) use pool::*;

Expand All @@ -83,12 +83,12 @@ pub(crate) use pool::*;
/// Protocol types needed for client operations
///
/// These types are used in the public API and must be imported for client usage:
/// - `ClientResult`: Response type from read operations
/// - `KvEntry`: Key-value pair returned by read operations
/// - `ReadConsistencyPolicy`: Consistency guarantees for reads
/// - `WriteCommand`: Write operation specifications
pub mod protocol {
pub use d_engine_proto::client::ClientResult;
pub use d_engine_proto::client::ReadConsistencyPolicy;
pub use d_engine_core::client::KvEntry;
pub use d_engine_core::config::ReadConsistencyPolicy;
pub use d_engine_proto::client::WatchEventType;
pub use d_engine_proto::client::WatchRequest;
pub use d_engine_proto::client::WatchResponse;
Expand Down
Loading
Loading