1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
//! Extensions for protobuf-generated client types
//!
//! This module provides additional methods for protobuf-generated types
//! that are not automatically generated by the protobuf compiler.
use bytes::Bytes;
use crate::client::ClientReadRequest;
use crate::client::ClientResponse;
use crate::client::ClientResult;
use crate::client::ReadConsistencyPolicy;
use crate::client::ReadResults;
use crate::client::WriteCommand;
use crate::client::WriteResult;
use crate::client::client_response::SuccessResult;
use crate::client::write_command;
use crate::error::ErrorCode;
use crate::error::ErrorMetadata;
impl ClientReadRequest {
/// Checks if the consistency_policy field is present in the request
///
/// Returns true if the client explicitly specified a consistency policy,
/// false if the field is absent (should use server default).
pub fn has_consistency_policy(&self) -> bool {
self.consistency_policy.is_some()
}
/// Gets the consistency policy value safely
///
/// Returns Some(policy) if present, None if field is absent.
/// Safer alternative that doesn't panic.
pub fn get_consistency_policy(&self) -> Option<ReadConsistencyPolicy> {
self.consistency_policy.and_then(|policy_i32| {
match policy_i32 {
x if x == ReadConsistencyPolicy::LeaseRead as i32 => {
Some(ReadConsistencyPolicy::LeaseRead)
}
x if x == ReadConsistencyPolicy::LinearizableRead as i32 => {
Some(ReadConsistencyPolicy::LinearizableRead)
}
_ => None, // Invalid value
}
})
}
}
impl WriteCommand {
/// Create write command for key-value pair
///
/// # Parameters
/// - `key`: Byte array for storage key
/// - `value`: Byte array to be stored
pub fn insert(
key: impl Into<Bytes>,
value: impl Into<Bytes>,
) -> Self {
let cmd = write_command::Insert {
key: key.into(),
value: value.into(),
ttl_secs: 0,
};
Self {
operation: Some(write_command::Operation::Insert(cmd)),
}
}
/// Create write command for key-value pair with TTL
///
/// # Parameters
/// - `key`: Byte array for storage key
/// - `value`: Byte array to be stored
/// - `ttl_secs`: Time-to-live in seconds
pub fn insert_with_ttl(
key: impl Into<Bytes>,
value: impl Into<Bytes>,
ttl_secs: u64,
) -> Self {
let cmd = write_command::Insert {
key: key.into(),
value: value.into(),
ttl_secs,
};
Self {
operation: Some(write_command::Operation::Insert(cmd)),
}
}
/// Create deletion command for specified key
///
/// # Parameters
/// - `key`: Byte array of key to delete
pub fn delete(key: impl Into<Bytes>) -> Self {
let cmd = write_command::Delete { key: key.into() };
Self {
operation: Some(write_command::Operation::Delete(cmd)),
}
}
/// Create compare-and-swap command
///
/// # Parameters
/// - `key`: Key to operate on
/// - `expected_value`: Expected current value (None means key must not exist)
/// - `new_value`: New value to set if comparison succeeds
pub fn compare_and_swap(
key: impl Into<Bytes>,
expected_value: Option<impl Into<Bytes>>,
new_value: impl Into<Bytes>,
) -> Self {
let cmd = write_command::CompareAndSwap {
key: key.into(),
expected_value: expected_value.map(|v| v.into()),
new_value: new_value.into(),
};
Self {
operation: Some(write_command::Operation::CompareAndSwap(cmd)),
}
}
}
impl ClientResponse {
/// Build success response for write operations
///
/// # Returns
/// Response with Success code and result=true
pub fn write_success() -> Self {
Self {
error: ErrorCode::Success as i32,
success_result: Some(SuccessResult::WriteResult(WriteResult { succeeded: true })),
metadata: None,
}
}
/// Build CAS success response (comparison succeeded, value updated)
///
/// # Returns
/// Response with Success code and result=true (CAS succeeded)
pub fn cas_success() -> Self {
Self {
error: ErrorCode::Success as i32,
success_result: Some(SuccessResult::WriteResult(WriteResult { succeeded: true })),
metadata: None,
}
}
/// Build CAS failure response (comparison failed, value NOT updated)
///
/// # Returns
/// Response with Success code and result=false (CAS failed due to mismatch)
pub fn cas_failure() -> Self {
Self {
error: ErrorCode::Success as i32,
success_result: Some(SuccessResult::WriteResult(WriteResult { succeeded: false })),
metadata: None,
}
}
/// Check if operation completed without error
///
/// # Returns
/// - `true` if no error occurred (operation executed)
/// - `false` if error occurred or invalid response
pub fn succeeded(&self) -> bool {
self.error == ErrorCode::Success as i32
&& matches!(self.success_result, Some(SuccessResult::WriteResult(_)))
}
/// Check if write operation succeeded with true result
///
/// Strict check for Put/Delete operations ensuring result is true.
/// For CAS, use pattern matching to distinguish true/false results.
///
/// # Returns
/// - `true` for successful Put/Delete (result must be true)
/// - `false` otherwise
pub fn is_write_success(&self) -> bool {
self.error == ErrorCode::Success as i32
&& matches!(
self.success_result,
Some(SuccessResult::WriteResult(WriteResult { succeeded: true }))
)
}
/// Build success response for read operations
///
/// # Parameters
/// - `results`: Vector of retrieved key-value pairs
pub fn read_results(results: Vec<ClientResult>) -> Self {
Self {
error: ErrorCode::Success as i32,
success_result: Some(SuccessResult::ReadData(ReadResults { results })),
metadata: None,
}
}
/// Build generic error response for any operation type
///
/// # Parameters
/// - `error_code`: Predefined client request error code
pub fn client_error(error_code: ErrorCode) -> Self {
Self {
error: error_code as i32,
success_result: None,
metadata: None,
}
}
/// Build NOT_LEADER error response with leader metadata
///
/// # Parameters
/// - `leader_id`: Optional leader node ID
/// - `leader_address`: Optional leader address
pub fn not_leader(
leader_id: Option<String>,
leader_address: Option<String>,
) -> Self {
let metadata = if leader_id.is_some() || leader_address.is_some() {
Some(ErrorMetadata {
retry_after_ms: None,
leader_id,
leader_address,
debug_message: None,
})
} else {
None
};
Self {
error: ErrorCode::NotLeader as i32,
success_result: None,
metadata,
}
}
/// Check if this response indicates the leader's term is outdated
pub fn is_term_outdated(&self) -> bool {
ErrorCode::try_from(self.error).map(|e| e.is_term_outdated()).unwrap_or(false)
}
/// Check if this response indicates a quorum timeout or failure to receive majority responses
pub fn is_quorum_timeout_or_failure(&self) -> bool {
ErrorCode::try_from(self.error)
.map(|e| e.is_quorum_timeout_or_failure())
.unwrap_or(false)
}
/// Check if this response indicates a failure to receive majority responses
pub fn is_propose_failure(&self) -> bool {
ErrorCode::try_from(self.error).map(|e| e.is_propose_failure()).unwrap_or(false)
}
/// Check if this response indicates a a retry required
pub fn is_retry_required(&self) -> bool {
ErrorCode::try_from(self.error).map(|e| e.is_retry_required()).unwrap_or(false)
}
}
impl ErrorCode {
/// Check if this error indicates the leader's term is outdated
pub fn is_term_outdated(&self) -> bool {
matches!(self, ErrorCode::TermOutdated)
}
/// Check if this error indicates a quorum timeout or failure to receive majority responses
pub fn is_quorum_timeout_or_failure(&self) -> bool {
matches!(
self,
ErrorCode::ConnectionTimeout | ErrorCode::ProposeFailed | ErrorCode::ClusterUnavailable
)
}
/// Check if this error indicates a failure to receive majority responses
pub fn is_propose_failure(&self) -> bool {
matches!(self, ErrorCode::ProposeFailed)
}
/// Check if this error indicates a retry required
pub fn is_retry_required(&self) -> bool {
matches!(self, ErrorCode::RetryRequired)
}
}