1use std::fmt;
2
3use dodb_core::{
4 ConditionExpectation, DocumentKey, Error, ObservedState, Revision, RevisionState,
5 TransactionCondition, TransactionConflict, TransactionMutation, TransactionRequest,
6};
7use dodb_service::{Document, Request, Response, TransactionOutcome};
8
9pub const MAGIC: [u8; 4] = *b"DODB";
10pub const PROTOCOL_VERSION: u16 = 1;
11pub const HEADER_SIZE: usize = 11;
12pub const REQUEST_MESSAGE_TYPE: u8 = 1;
13pub const RESPONSE_MESSAGE_TYPE: u8 = 2;
14pub const MAX_STORAGE_VALUE_SIZE: usize = 64 * 1024 * 1024;
15pub const MAX_STORAGE_KEY_COMPONENT_SIZE: usize = 3_990;
16pub const MAX_STORAGE_ENCODED_KEY_SIZE: usize = 3_992;
17
18#[derive(Clone, Copy, Debug, Eq, PartialEq)]
19pub struct ProtocolLimits {
20 pub max_request_frame_size: usize,
21 pub max_response_frame_size: usize,
22 pub max_key_component_size: usize,
23 pub max_encoded_key_size: usize,
24 pub max_value_size: usize,
25 pub max_keys: usize,
26 pub max_conditions: usize,
27 pub max_mutations: usize,
28 pub max_query_limit: usize,
29 pub max_scan_limit: usize,
30 pub max_error_detail_size: usize,
31}
32
33impl Default for ProtocolLimits {
34 fn default() -> Self {
35 Self {
36 max_request_frame_size: 68 * 1024 * 1024,
37 max_response_frame_size: 68 * 1024 * 1024,
38 max_key_component_size: MAX_STORAGE_KEY_COMPONENT_SIZE,
39 max_encoded_key_size: MAX_STORAGE_ENCODED_KEY_SIZE,
40 max_value_size: MAX_STORAGE_VALUE_SIZE,
41 max_keys: 4_096,
42 max_conditions: 256,
43 max_mutations: 256,
44 max_query_limit: 4_096,
45 max_scan_limit: 4_096,
46 max_error_detail_size: 4_096,
47 }
48 }
49}
50
51impl ProtocolLimits {
52 pub fn validate(&self) -> Result<(), ProtocolError> {
53 if self.max_request_frame_size < HEADER_SIZE
54 || self.max_response_frame_size < HEADER_SIZE
55 || self.max_key_component_size == 0
56 || self.max_encoded_key_size == 0
57 || self.max_value_size == 0
58 || self.max_keys == 0
59 || self.max_conditions == 0
60 || self.max_mutations == 0
61 || self.max_query_limit == 0
62 || self.max_scan_limit == 0
63 || self.max_error_detail_size == 0
64 {
65 return Err(ProtocolError::InvalidLimits);
66 }
67 if self.max_key_component_size > MAX_STORAGE_KEY_COMPONENT_SIZE
68 || self.max_encoded_key_size > MAX_STORAGE_ENCODED_KEY_SIZE
69 || self.max_value_size > MAX_STORAGE_VALUE_SIZE
70 {
71 return Err(ProtocolError::InvalidLimits);
72 }
73 Ok(())
74 }
75}
76
77#[derive(Clone, Copy, Debug, Eq, PartialEq)]
78pub struct FrameHeader {
79 pub version: u16,
80 pub message_type: u8,
81 pub payload_length: usize,
82}
83
84#[derive(Clone, Debug, Eq, PartialEq)]
85pub enum ProtocolError {
86 InvalidLimits,
87 TruncatedHeader,
88 InvalidMagic,
89 UnsupportedVersion(u16),
90 UnknownMessageType(u8),
91 TruncatedPayload,
92 TrailingBytes,
93 PayloadTooLarge { length: usize, maximum: usize },
94 KeyTooLarge { length: usize, maximum: usize },
95 LengthOverflow,
96 InvalidOpcode(u8),
97 InvalidStatus(u8),
98 InvalidFlag(u8),
99 InvalidResponseType { expected: u8, actual: u8 },
100 InvalidErrorKind(u8),
101 InvalidExpectation(u8),
102 InvalidRevisionState(u8),
103 InvalidUtf8,
104}
105
106impl fmt::Display for ProtocolError {
107 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
108 match self {
109 Self::InvalidLimits => write!(formatter, "protocol limits are invalid"),
110 Self::TruncatedHeader => write!(formatter, "truncated frame header"),
111 Self::InvalidMagic => write!(formatter, "invalid frame magic"),
112 Self::UnsupportedVersion(version) => {
113 write!(formatter, "unsupported protocol version {version}")
114 }
115 Self::UnknownMessageType(message_type) => {
116 write!(formatter, "unknown message type {message_type}")
117 }
118 Self::TruncatedPayload => write!(formatter, "truncated frame payload"),
119 Self::TrailingBytes => write!(formatter, "unexpected trailing bytes"),
120 Self::PayloadTooLarge { length, maximum } => {
121 write!(
122 formatter,
123 "payload length {length} exceeds maximum {maximum}"
124 )
125 }
126 Self::KeyTooLarge { length, maximum } => {
127 write!(
128 formatter,
129 "encoded key length {length} exceeds maximum {maximum}"
130 )
131 }
132 Self::LengthOverflow => write!(formatter, "length conversion overflow"),
133 Self::InvalidOpcode(opcode) => write!(formatter, "invalid operation opcode {opcode}"),
134 Self::InvalidStatus(status) => write!(formatter, "invalid response status {status}"),
135 Self::InvalidFlag(flag) => write!(formatter, "invalid boolean flag {flag}"),
136 Self::InvalidResponseType { expected, actual } => write!(
137 formatter,
138 "response type {actual} does not match expected type {expected}"
139 ),
140 Self::InvalidErrorKind(kind) => {
141 write!(formatter, "invalid application error kind {kind}")
142 }
143 Self::InvalidExpectation(expectation) => {
144 write!(formatter, "invalid transaction expectation {expectation}")
145 }
146 Self::InvalidRevisionState(state) => {
147 write!(formatter, "invalid revision state {state}")
148 }
149 Self::InvalidUtf8 => write!(formatter, "invalid UTF-8 error detail"),
150 }
151 }
152}
153
154impl std::error::Error for ProtocolError {}
155
156#[derive(Clone, Copy, Debug, Eq, PartialEq)]
157pub enum ApplicationErrorKind {
158 InvalidRequest,
159 Overloaded,
160 Conflict,
161 ResponseTooLarge,
162 StorageFailure,
163 Corruption,
164 DurabilityFailure,
165 Internal,
166 UnsupportedProtocol,
167}
168
169#[derive(Clone, Copy, Debug, Eq, PartialEq)]
170pub enum MutationOutcome {
171 NotApplicable,
172 NotApplied,
173 Unknown,
174}
175
176#[derive(Clone, Debug, Eq, PartialEq)]
177pub struct ApplicationError {
178 pub kind: ApplicationErrorKind,
179 pub detail: String,
180 pub mutation_outcome: MutationOutcome,
181 pub conflict: Option<ConflictDetails>,
182}
183
184#[derive(Clone, Debug, Eq, PartialEq)]
185pub struct ConflictDetails {
186 pub key: DocumentKey,
187 pub expected: ConditionExpectation,
188 pub actual: ObservedState,
189}
190
191impl ApplicationError {
192 pub fn invalid_request(detail: impl Into<String>) -> Self {
193 Self {
194 kind: ApplicationErrorKind::InvalidRequest,
195 detail: detail.into(),
196 mutation_outcome: MutationOutcome::NotApplied,
197 conflict: None,
198 }
199 }
200
201 pub fn from_core(error: &Error) -> Self {
202 match error {
203 Error::InvalidInput(detail) | Error::InvalidRequest(detail) => Self {
204 kind: ApplicationErrorKind::InvalidRequest,
205 detail: detail.clone(),
206 mutation_outcome: MutationOutcome::NotApplied,
207 conflict: None,
208 },
209 Error::Overloaded(detail) => Self {
210 kind: ApplicationErrorKind::Overloaded,
211 detail: detail.clone(),
212 mutation_outcome: MutationOutcome::NotApplied,
213 conflict: None,
214 },
215 Error::Conflict(conflict) => Self {
216 kind: ApplicationErrorKind::Conflict,
217 detail: "transaction condition conflict".to_owned(),
218 mutation_outcome: MutationOutcome::NotApplied,
219 conflict: Some(ConflictDetails::from(conflict)),
220 },
221 Error::ResponseTooLarge(detail) => Self {
222 kind: ApplicationErrorKind::ResponseTooLarge,
223 detail: detail.clone(),
224 mutation_outcome: MutationOutcome::NotApplied,
225 conflict: None,
226 },
227 Error::Corruption(detail) => Self {
228 kind: ApplicationErrorKind::Corruption,
229 detail: detail.clone(),
230 mutation_outcome: MutationOutcome::Unknown,
231 conflict: None,
232 },
233 Error::Io(error) => Self {
234 kind: ApplicationErrorKind::StorageFailure,
235 detail: error.to_string(),
236 mutation_outcome: MutationOutcome::Unknown,
237 conflict: None,
238 },
239 Error::UnsupportedFormat(detail)
240 | Error::RecoveryFailure(detail)
241 | Error::CheckpointFailure(detail) => Self {
242 kind: ApplicationErrorKind::StorageFailure,
243 detail: detail.clone(),
244 mutation_outcome: MutationOutcome::Unknown,
245 conflict: None,
246 },
247 Error::DurabilityFailure(detail) => Self {
248 kind: ApplicationErrorKind::DurabilityFailure,
249 detail: detail.clone(),
250 mutation_outcome: MutationOutcome::Unknown,
251 conflict: None,
252 },
253 Error::InternalInvariantViolation(detail) => Self {
254 kind: ApplicationErrorKind::Internal,
255 detail: detail.clone(),
256 mutation_outcome: MutationOutcome::Unknown,
257 conflict: None,
258 },
259 }
260 }
261
262 pub fn from_core_for_request(error: &Error, is_mutation: bool) -> Self {
263 let mut application_error = Self::from_core(error);
264 if !is_mutation {
265 application_error.mutation_outcome = MutationOutcome::NotApplicable;
266 }
267 application_error
268 }
269
270 pub fn from_protocol(error: &ProtocolError) -> Self {
271 let kind = match error {
272 ProtocolError::UnsupportedVersion(_) | ProtocolError::UnknownMessageType(_) => {
273 ApplicationErrorKind::UnsupportedProtocol
274 }
275 _ => ApplicationErrorKind::InvalidRequest,
276 };
277 Self {
278 kind,
279 detail: error.to_string(),
280 mutation_outcome: MutationOutcome::NotApplied,
281 conflict: None,
282 }
283 }
284}
285
286impl From<&TransactionConflict> for ConflictDetails {
287 fn from(conflict: &TransactionConflict) -> Self {
288 Self {
289 key: conflict.key.clone(),
290 expected: conflict.expected,
291 actual: conflict.actual,
292 }
293 }
294}
295
296#[derive(Clone, Debug, Eq, PartialEq)]
297pub enum ResponseEnvelope {
298 Success(Response),
299 Error(ApplicationError),
300}
301
302pub fn encode_request(
303 tenant: dodb_core::TenantId,
304 request: &Request,
305 limits: ProtocolLimits,
306) -> Result<Vec<u8>, ProtocolError> {
307 limits.validate()?;
308 let mut payload = Writer::new();
309 payload.u64(tenant.get());
310 encode_request_payload(&mut payload, request, limits)?;
311 encode_frame(
312 REQUEST_MESSAGE_TYPE,
313 payload.finish(),
314 limits.max_request_frame_size,
315 )
316}
317
318pub fn decode_request_frame(
319 frame: &[u8],
320 limits: ProtocolLimits,
321) -> Result<(dodb_core::TenantId, Request), ProtocolError> {
322 let header = decode_header(
323 frame
324 .get(..HEADER_SIZE)
325 .ok_or(ProtocolError::TruncatedHeader)?,
326 )?;
327 if frame.len() != HEADER_SIZE + header.payload_length {
328 return Err(if frame.len() < HEADER_SIZE + header.payload_length {
329 ProtocolError::TruncatedPayload
330 } else {
331 ProtocolError::TrailingBytes
332 });
333 }
334 decode_request_parts(header, &frame[HEADER_SIZE..], limits)
335}
336
337pub fn decode_request_parts(
338 header: FrameHeader,
339 payload: &[u8],
340 limits: ProtocolLimits,
341) -> Result<(dodb_core::TenantId, Request), ProtocolError> {
342 limits.validate()?;
343 validate_payload_length(header, payload, limits.max_request_frame_size)?;
344 if header.message_type != REQUEST_MESSAGE_TYPE {
345 return Err(ProtocolError::UnknownMessageType(header.message_type));
346 }
347 let mut reader = Reader::new(payload);
348 let tenant = dodb_core::TenantId::new(reader.u64()?);
349 let request = decode_request_payload(&mut reader, limits)?;
350 reader
351 .finish()?
352 .then_some((tenant, request))
353 .ok_or(ProtocolError::TrailingBytes)
354}
355
356pub fn encode_response(
357 response: &ResponseEnvelope,
358 limits: ProtocolLimits,
359) -> Result<Vec<u8>, ProtocolError> {
360 limits.validate()?;
361 let mut payload = Writer::new();
362 match response {
363 ResponseEnvelope::Success(response) => {
364 payload.u8(0);
365 payload.u8(response_opcode(response));
366 encode_response_payload(&mut payload, response, limits)?;
367 }
368 ResponseEnvelope::Error(error) => {
369 payload.u8(1);
370 encode_application_error(&mut payload, error, limits)?;
371 }
372 }
373 encode_frame(
374 RESPONSE_MESSAGE_TYPE,
375 payload.finish(),
376 limits.max_response_frame_size,
377 )
378}
379
380pub fn decode_response_frame(
381 frame: &[u8],
382 expected_response_type: Option<u8>,
383 limits: ProtocolLimits,
384) -> Result<ResponseEnvelope, ProtocolError> {
385 let header = decode_header(
386 frame
387 .get(..HEADER_SIZE)
388 .ok_or(ProtocolError::TruncatedHeader)?,
389 )?;
390 if frame.len() != HEADER_SIZE + header.payload_length {
391 return Err(if frame.len() < HEADER_SIZE + header.payload_length {
392 ProtocolError::TruncatedPayload
393 } else {
394 ProtocolError::TrailingBytes
395 });
396 }
397 decode_response_parts(
398 header,
399 &frame[HEADER_SIZE..],
400 expected_response_type,
401 limits,
402 )
403}
404
405pub fn decode_response_parts(
406 header: FrameHeader,
407 payload: &[u8],
408 expected_response_type: Option<u8>,
409 limits: ProtocolLimits,
410) -> Result<ResponseEnvelope, ProtocolError> {
411 limits.validate()?;
412 validate_payload_length(header, payload, limits.max_response_frame_size)?;
413 if header.message_type != RESPONSE_MESSAGE_TYPE {
414 return Err(ProtocolError::UnknownMessageType(header.message_type));
415 }
416 let mut reader = Reader::new(payload);
417 let status = reader.u8()?;
418 let response = match status {
419 0 => {
420 let actual = reader.u8()?;
421 if let Some(expected) = expected_response_type
422 && actual != expected
423 {
424 return Err(ProtocolError::InvalidResponseType { expected, actual });
425 }
426 ResponseEnvelope::Success(decode_response_payload(&mut reader, actual, limits)?)
427 }
428 1 => ResponseEnvelope::Error(decode_application_error(&mut reader, limits)?),
429 other => return Err(ProtocolError::InvalidStatus(other)),
430 };
431 reader
432 .finish()?
433 .then_some(response)
434 .ok_or(ProtocolError::TrailingBytes)
435}
436
437pub fn request_opcode(request: &Request) -> u8 {
438 match request {
439 Request::Get { .. } => 1,
440 Request::Put { .. } => 2,
441 Request::Delete { .. } => 3,
442 Request::Query { .. } => 4,
443 Request::Scan { .. } => 5,
444 Request::Transact { .. } => 8,
445 }
446}
447
448pub fn response_opcode(response: &Response) -> u8 {
449 match response {
450 Response::Get(_) => 1,
451 Response::Put(_) => 2,
452 Response::Delete(_) => 3,
453 Response::Query(_) => 4,
454 Response::Scan(_) => 5,
455 Response::Transact(_) => 8,
456 }
457}
458
459pub fn decode_header(bytes: &[u8]) -> Result<FrameHeader, ProtocolError> {
460 if bytes.len() < HEADER_SIZE {
461 return Err(ProtocolError::TruncatedHeader);
462 }
463 if bytes[..4] != MAGIC {
464 return Err(ProtocolError::InvalidMagic);
465 }
466 let version = u16::from_be_bytes([bytes[4], bytes[5]]);
467 if version != PROTOCOL_VERSION {
468 return Err(ProtocolError::UnsupportedVersion(version));
469 }
470 let message_type = bytes[6];
471 if message_type != REQUEST_MESSAGE_TYPE && message_type != RESPONSE_MESSAGE_TYPE {
472 return Err(ProtocolError::UnknownMessageType(message_type));
473 }
474 let payload_length = u32::from_be_bytes([bytes[7], bytes[8], bytes[9], bytes[10]]) as usize;
475 Ok(FrameHeader {
476 version,
477 message_type,
478 payload_length,
479 })
480}
481
482fn encode_frame(
483 message_type: u8,
484 payload: Vec<u8>,
485 maximum_frame_size: usize,
486) -> Result<Vec<u8>, ProtocolError> {
487 let frame_length = HEADER_SIZE
488 .checked_add(payload.len())
489 .ok_or(ProtocolError::LengthOverflow)?;
490 if frame_length > maximum_frame_size {
491 return Err(ProtocolError::PayloadTooLarge {
492 length: frame_length,
493 maximum: maximum_frame_size,
494 });
495 }
496 let payload_length = u32::try_from(payload.len()).map_err(|_| ProtocolError::LengthOverflow)?;
497 let mut frame = Vec::with_capacity(frame_length);
498 frame.extend_from_slice(&MAGIC);
499 frame.extend_from_slice(&PROTOCOL_VERSION.to_be_bytes());
500 frame.push(message_type);
501 frame.extend_from_slice(&payload_length.to_be_bytes());
502 frame.extend_from_slice(&payload);
503 Ok(frame)
504}
505
506fn validate_payload_length(
507 header: FrameHeader,
508 payload: &[u8],
509 maximum_frame_size: usize,
510) -> Result<(), ProtocolError> {
511 let frame_length = HEADER_SIZE
512 .checked_add(header.payload_length)
513 .ok_or(ProtocolError::LengthOverflow)?;
514 if frame_length > maximum_frame_size {
515 return Err(ProtocolError::PayloadTooLarge {
516 length: frame_length,
517 maximum: maximum_frame_size,
518 });
519 }
520 if payload.len() != header.payload_length {
521 return Err(if payload.len() < header.payload_length {
522 ProtocolError::TruncatedPayload
523 } else {
524 ProtocolError::TrailingBytes
525 });
526 }
527 Ok(())
528}
529
530fn encode_request_payload(
531 writer: &mut Writer,
532 request: &Request,
533 limits: ProtocolLimits,
534) -> Result<(), ProtocolError> {
535 writer.u8(request_opcode(request));
536 match request {
537 Request::Get { key } => encode_key(writer, key, limits)?,
538 Request::Put { key, value } => {
539 encode_key(writer, key, limits)?;
540 writer.bytes(value, limits.max_value_size)?;
541 }
542 Request::Delete { key } => encode_key(writer, key, limits)?,
543 Request::Query {
544 pk,
545 exclusive_after_sk,
546 limit,
547 } => {
548 validate_key_parts(
549 pk.as_bytes(),
550 exclusive_after_sk
551 .as_ref()
552 .map_or(&[][..], dodb_core::SortKey::as_bytes),
553 limits,
554 )?;
555 encode_bytes(writer, pk.as_bytes(), limits.max_key_component_size)?;
556 encode_optional_bytes(
557 writer,
558 exclusive_after_sk
559 .as_ref()
560 .map(dodb_core::SortKey::as_bytes),
561 limits.max_key_component_size,
562 )?;
563 encode_limit(writer, *limit, limits.max_query_limit)?;
564 }
565 Request::Scan {
566 exclusive_after_key,
567 limit,
568 } => {
569 encode_optional_key(writer, exclusive_after_key.as_ref(), limits)?;
570 encode_limit(writer, *limit, limits.max_scan_limit)?;
571 }
572 Request::Transact { request } => {
573 encode_conditions(writer, &request.conditions, limits)?;
574 encode_mutations(writer, &request.mutations, limits)?;
575 }
576 }
577 Ok(())
578}
579
580fn decode_request_payload(
581 reader: &mut Reader<'_>,
582 limits: ProtocolLimits,
583) -> Result<Request, ProtocolError> {
584 let opcode = reader.u8()?;
585 match opcode {
586 1 => Ok(Request::Get {
587 key: decode_key(reader, limits)?,
588 }),
589 2 => Ok(Request::Put {
590 key: decode_key(reader, limits)?,
591 value: reader.bytes(limits.max_value_size)?,
592 }),
593 3 => Ok(Request::Delete {
594 key: decode_key(reader, limits)?,
595 }),
596 4 => {
597 let pk_bytes = reader.bytes(limits.max_key_component_size)?;
598 let exclusive_after_sk_bytes = reader.optional_bytes(limits.max_key_component_size)?;
599 validate_key_parts(
600 &pk_bytes,
601 exclusive_after_sk_bytes.as_deref().unwrap_or(&[]),
602 limits,
603 )?;
604 let pk = dodb_core::PrimaryKey::new(pk_bytes);
605 let exclusive_after_sk = exclusive_after_sk_bytes.map(dodb_core::SortKey::new);
606 let limit = reader.limit(limits.max_query_limit)?;
607 Ok(Request::Query {
608 pk,
609 exclusive_after_sk,
610 limit,
611 })
612 }
613 5 => Ok(Request::Scan {
614 exclusive_after_key: decode_optional_key(reader, limits)?,
615 limit: reader.limit(limits.max_scan_limit)?,
616 }),
617 8 => Ok(Request::Transact {
618 request: TransactionRequest::new(
619 decode_conditions(reader, limits)?,
620 decode_mutations(reader, limits)?,
621 ),
622 }),
623 other => Err(ProtocolError::InvalidOpcode(other)),
624 }
625}
626
627fn encode_response_payload(
628 writer: &mut Writer,
629 response: &Response,
630 limits: ProtocolLimits,
631) -> Result<(), ProtocolError> {
632 match response {
633 Response::Get(state) => encode_revision_state(writer, state, limits)?,
634 Response::Put(revision) | Response::Delete(revision) => writer.u64(revision.get()),
635 Response::Query(documents) | Response::Scan(documents) => {
636 encode_documents(writer, documents, limits)?
637 }
638 Response::Transact(outcome) => encode_transaction_outcome(writer, *outcome),
639 }
640 Ok(())
641}
642
643fn decode_response_payload(
644 reader: &mut Reader<'_>,
645 opcode: u8,
646 limits: ProtocolLimits,
647) -> Result<Response, ProtocolError> {
648 match opcode {
649 1 => Ok(Response::Get(decode_revision_state(reader, limits)?)),
650 2 => Ok(Response::Put(Revision::new(reader.u64()?))),
651 3 => Ok(Response::Delete(Revision::new(reader.u64()?))),
652 4 => Ok(Response::Query(decode_documents(reader, limits)?)),
653 5 => Ok(Response::Scan(decode_documents(reader, limits)?)),
654 8 => Ok(Response::Transact(decode_transaction_outcome(reader)?)),
655 other => Err(ProtocolError::InvalidOpcode(other)),
656 }
657}
658
659fn encode_transaction_outcome(writer: &mut Writer, outcome: TransactionOutcome) {
660 match outcome.commit_lsn {
661 Some(commit_lsn) => {
662 writer.u8(1);
663 writer.u64(commit_lsn.get());
664 }
665 None => writer.u8(0),
666 }
667}
668
669fn decode_transaction_outcome(
670 reader: &mut Reader<'_>,
671) -> Result<TransactionOutcome, ProtocolError> {
672 match reader.u8()? {
673 0 => Ok(TransactionOutcome::conditions_satisfied()),
674 1 => Ok(TransactionOutcome::committed(dodb_core::Lsn::new(
675 reader.u64()?,
676 ))),
677 other => Err(ProtocolError::InvalidFlag(other)),
678 }
679}
680
681fn encode_documents(
682 writer: &mut Writer,
683 documents: &[Document],
684 limits: ProtocolLimits,
685) -> Result<(), ProtocolError> {
686 encode_count(writer, documents.len(), limits.max_scan_limit)?;
687 for document in documents {
688 encode_key(writer, &document.key, limits)?;
689 writer.bytes(&document.value, limits.max_value_size)?;
690 writer.u64(document.revision.get());
691 }
692 Ok(())
693}
694
695fn decode_documents(
696 reader: &mut Reader<'_>,
697 limits: ProtocolLimits,
698) -> Result<Vec<Document>, ProtocolError> {
699 let count = reader.count(limits.max_scan_limit)?;
700 let mut documents = Vec::with_capacity(count);
701 for _ in 0..count {
702 documents.push(Document {
703 key: decode_key(reader, limits)?,
704 value: reader.bytes(limits.max_value_size)?,
705 revision: Revision::new(reader.u64()?),
706 });
707 }
708 Ok(documents)
709}
710
711fn encode_revision_state(
712 writer: &mut Writer,
713 state: &RevisionState,
714 limits: ProtocolLimits,
715) -> Result<(), ProtocolError> {
716 match state {
717 RevisionState::Present { value, revision } => {
718 writer.u8(0);
719 writer.bytes(value, limits.max_value_size)?;
720 writer.u64(revision.get());
721 }
722 RevisionState::Missing { revision } => {
723 writer.u8(1);
724 writer.u64(revision.get());
725 }
726 }
727 Ok(())
728}
729
730fn decode_revision_state(
731 reader: &mut Reader<'_>,
732 limits: ProtocolLimits,
733) -> Result<RevisionState, ProtocolError> {
734 match reader.u8()? {
735 0 => Ok(RevisionState::present(
736 reader.bytes(limits.max_value_size)?,
737 Revision::new(reader.u64()?),
738 )),
739 1 => Ok(RevisionState::missing(Revision::new(reader.u64()?))),
740 other => Err(ProtocolError::InvalidRevisionState(other)),
741 }
742}
743
744fn encode_observed_state(writer: &mut Writer, state: ObservedState) {
745 match state {
746 ObservedState::Present { revision } => {
747 writer.u8(0);
748 writer.u64(revision.get());
749 }
750 ObservedState::Missing { revision } => {
751 writer.u8(1);
752 writer.u64(revision.get());
753 }
754 }
755}
756
757fn decode_observed_state(reader: &mut Reader<'_>) -> Result<ObservedState, ProtocolError> {
758 match reader.u8()? {
759 0 => Ok(ObservedState::present(Revision::new(reader.u64()?))),
760 1 => Ok(ObservedState::missing(Revision::new(reader.u64()?))),
761 other => Err(ProtocolError::InvalidRevisionState(other)),
762 }
763}
764
765fn encode_conditions(
766 writer: &mut Writer,
767 conditions: &[TransactionCondition],
768 limits: ProtocolLimits,
769) -> Result<(), ProtocolError> {
770 encode_count(writer, conditions.len(), limits.max_conditions)?;
771 for condition in conditions {
772 match condition {
773 TransactionCondition::RevisionEquals {
774 key,
775 expected_revision,
776 } => {
777 writer.u8(0);
778 encode_key(writer, key, limits)?;
779 writer.u64(expected_revision.get());
780 }
781 TransactionCondition::Exists { key } => {
782 writer.u8(1);
783 encode_key(writer, key, limits)?;
784 }
785 TransactionCondition::NotExists { key } => {
786 writer.u8(2);
787 encode_key(writer, key, limits)?;
788 }
789 }
790 }
791 Ok(())
792}
793
794fn decode_conditions(
795 reader: &mut Reader<'_>,
796 limits: ProtocolLimits,
797) -> Result<Vec<TransactionCondition>, ProtocolError> {
798 let count = reader.count(limits.max_conditions)?;
799 let mut conditions = Vec::with_capacity(count);
800 for _ in 0..count {
801 let kind = reader.u8()?;
802 let key = decode_key(reader, limits)?;
803 conditions.push(match kind {
804 0 => TransactionCondition::RevisionEquals {
805 key,
806 expected_revision: Revision::new(reader.u64()?),
807 },
808 1 => TransactionCondition::Exists { key },
809 2 => TransactionCondition::NotExists { key },
810 other => return Err(ProtocolError::InvalidExpectation(other)),
811 });
812 }
813 Ok(conditions)
814}
815
816fn encode_mutations(
817 writer: &mut Writer,
818 mutations: &[TransactionMutation],
819 limits: ProtocolLimits,
820) -> Result<(), ProtocolError> {
821 encode_count(writer, mutations.len(), limits.max_mutations)?;
822 for mutation in mutations {
823 match mutation {
824 TransactionMutation::Put { key, value } => {
825 writer.u8(0);
826 encode_key(writer, key, limits)?;
827 writer.bytes(value, limits.max_value_size)?;
828 }
829 TransactionMutation::Delete { key } => {
830 writer.u8(1);
831 encode_key(writer, key, limits)?;
832 }
833 }
834 }
835 Ok(())
836}
837
838fn decode_mutations(
839 reader: &mut Reader<'_>,
840 limits: ProtocolLimits,
841) -> Result<Vec<TransactionMutation>, ProtocolError> {
842 let count = reader.count(limits.max_mutations)?;
843 let mut mutations = Vec::with_capacity(count);
844 for _ in 0..count {
845 let kind = reader.u8()?;
846 let key = decode_key(reader, limits)?;
847 mutations.push(match kind {
848 0 => TransactionMutation::Put {
849 key,
850 value: reader.bytes(limits.max_value_size)?,
851 },
852 1 => TransactionMutation::Delete { key },
853 other => return Err(ProtocolError::InvalidOpcode(other)),
854 });
855 }
856 Ok(mutations)
857}
858
859fn encode_key(
860 writer: &mut Writer,
861 key: &DocumentKey,
862 limits: ProtocolLimits,
863) -> Result<(), ProtocolError> {
864 validate_key(key, limits)?;
865 encode_bytes(writer, key.pk.as_bytes(), limits.max_key_component_size)?;
866 encode_bytes(writer, key.sk.as_bytes(), limits.max_key_component_size)
867}
868
869fn decode_key(
870 reader: &mut Reader<'_>,
871 limits: ProtocolLimits,
872) -> Result<DocumentKey, ProtocolError> {
873 let key = DocumentKey::new(
874 reader.bytes(limits.max_key_component_size)?,
875 reader.bytes(limits.max_key_component_size)?,
876 );
877 validate_key(&key, limits)?;
878 Ok(key)
879}
880
881fn validate_key(key: &DocumentKey, limits: ProtocolLimits) -> Result<(), ProtocolError> {
882 validate_key_parts(key.pk.as_bytes(), key.sk.as_bytes(), limits)
883}
884
885fn validate_key_parts(pk: &[u8], sk: &[u8], limits: ProtocolLimits) -> Result<(), ProtocolError> {
886 let key = DocumentKey::new(pk.to_vec(), sk.to_vec());
887 let encoded_length = key.encoded_len();
888 if encoded_length > limits.max_encoded_key_size {
889 return Err(ProtocolError::KeyTooLarge {
890 length: encoded_length,
891 maximum: limits.max_encoded_key_size,
892 });
893 }
894 Ok(())
895}
896
897fn encode_optional_key(
898 writer: &mut Writer,
899 key: Option<&DocumentKey>,
900 limits: ProtocolLimits,
901) -> Result<(), ProtocolError> {
902 match key {
903 Some(key) => {
904 writer.u8(1);
905 encode_key(writer, key, limits)?;
906 }
907 None => writer.u8(0),
908 }
909 Ok(())
910}
911
912fn decode_optional_key(
913 reader: &mut Reader<'_>,
914 limits: ProtocolLimits,
915) -> Result<Option<DocumentKey>, ProtocolError> {
916 match reader.flag()? {
917 false => Ok(None),
918 true => Ok(Some(decode_key(reader, limits)?)),
919 }
920}
921
922fn encode_optional_bytes(
923 writer: &mut Writer,
924 bytes: Option<&[u8]>,
925 maximum: usize,
926) -> Result<(), ProtocolError> {
927 match bytes {
928 Some(bytes) => {
929 writer.u8(1);
930 encode_bytes(writer, bytes, maximum)?;
931 }
932 None => writer.u8(0),
933 }
934 Ok(())
935}
936
937fn encode_limit(writer: &mut Writer, limit: usize, maximum: usize) -> Result<(), ProtocolError> {
938 if limit > maximum {
939 return Err(ProtocolError::PayloadTooLarge {
940 length: limit,
941 maximum,
942 });
943 }
944 let limit = u32::try_from(limit).map_err(|_| ProtocolError::LengthOverflow)?;
945 writer.u32(limit);
946 Ok(())
947}
948
949fn encode_count(writer: &mut Writer, count: usize, maximum: usize) -> Result<(), ProtocolError> {
950 if count > maximum {
951 return Err(ProtocolError::PayloadTooLarge {
952 length: count,
953 maximum,
954 });
955 }
956 writer.u32(u32::try_from(count).map_err(|_| ProtocolError::LengthOverflow)?);
957 Ok(())
958}
959
960fn encode_bytes(writer: &mut Writer, bytes: &[u8], maximum: usize) -> Result<(), ProtocolError> {
961 if bytes.len() > maximum {
962 return Err(ProtocolError::PayloadTooLarge {
963 length: bytes.len(),
964 maximum,
965 });
966 }
967 writer.bytes_raw(bytes)
968}
969
970fn encode_application_error(
971 writer: &mut Writer,
972 error: &ApplicationError,
973 limits: ProtocolLimits,
974) -> Result<(), ProtocolError> {
975 writer.u8(application_error_kind_code(error.kind));
976 let detail = error
977 .detail
978 .chars()
979 .take(limits.max_error_detail_size)
980 .collect::<String>();
981 encode_bytes(writer, detail.as_bytes(), limits.max_error_detail_size)?;
982 writer.u8(mutation_outcome_code(error.mutation_outcome));
983 match &error.conflict {
984 Some(conflict) => {
985 writer.u8(1);
986 encode_conflict(writer, conflict, limits)?;
987 }
988 None => writer.u8(0),
989 }
990 Ok(())
991}
992
993fn decode_application_error(
994 reader: &mut Reader<'_>,
995 limits: ProtocolLimits,
996) -> Result<ApplicationError, ProtocolError> {
997 let kind = application_error_kind(reader.u8()?)?;
998 let detail = String::from_utf8(reader.bytes(limits.max_error_detail_size)?)
999 .map_err(|_| ProtocolError::InvalidUtf8)?;
1000 let mutation_outcome = mutation_outcome(reader.u8()?)?;
1001 let conflict = match reader.flag()? {
1002 true => Some(decode_conflict(reader, limits)?),
1003 false => None,
1004 };
1005 Ok(ApplicationError {
1006 kind,
1007 detail,
1008 mutation_outcome,
1009 conflict,
1010 })
1011}
1012
1013fn encode_conflict(
1014 writer: &mut Writer,
1015 conflict: &ConflictDetails,
1016 limits: ProtocolLimits,
1017) -> Result<(), ProtocolError> {
1018 encode_key(writer, &conflict.key, limits)?;
1019 match conflict.expected {
1020 ConditionExpectation::RevisionEquals(revision) => {
1021 writer.u8(0);
1022 writer.u64(revision.get());
1023 }
1024 ConditionExpectation::Exists => writer.u8(1),
1025 ConditionExpectation::NotExists => writer.u8(2),
1026 }
1027 encode_observed_state(writer, conflict.actual);
1028 Ok(())
1029}
1030
1031fn decode_conflict(
1032 reader: &mut Reader<'_>,
1033 limits: ProtocolLimits,
1034) -> Result<ConflictDetails, ProtocolError> {
1035 let key = decode_key(reader, limits)?;
1036 let expected = match reader.u8()? {
1037 0 => ConditionExpectation::RevisionEquals(Revision::new(reader.u64()?)),
1038 1 => ConditionExpectation::Exists,
1039 2 => ConditionExpectation::NotExists,
1040 other => return Err(ProtocolError::InvalidExpectation(other)),
1041 };
1042 Ok(ConflictDetails {
1043 key,
1044 expected,
1045 actual: decode_observed_state(reader)?,
1046 })
1047}
1048
1049fn application_error_kind_code(kind: ApplicationErrorKind) -> u8 {
1050 match kind {
1051 ApplicationErrorKind::InvalidRequest => 0,
1052 ApplicationErrorKind::Overloaded => 1,
1053 ApplicationErrorKind::Conflict => 2,
1054 ApplicationErrorKind::StorageFailure => 3,
1055 ApplicationErrorKind::Corruption => 4,
1056 ApplicationErrorKind::DurabilityFailure => 5,
1057 ApplicationErrorKind::Internal => 6,
1058 ApplicationErrorKind::UnsupportedProtocol => 7,
1059 ApplicationErrorKind::ResponseTooLarge => 8,
1060 }
1061}
1062
1063fn application_error_kind(code: u8) -> Result<ApplicationErrorKind, ProtocolError> {
1064 match code {
1065 0 => Ok(ApplicationErrorKind::InvalidRequest),
1066 1 => Ok(ApplicationErrorKind::Overloaded),
1067 2 => Ok(ApplicationErrorKind::Conflict),
1068 3 => Ok(ApplicationErrorKind::StorageFailure),
1069 4 => Ok(ApplicationErrorKind::Corruption),
1070 5 => Ok(ApplicationErrorKind::DurabilityFailure),
1071 6 => Ok(ApplicationErrorKind::Internal),
1072 7 => Ok(ApplicationErrorKind::UnsupportedProtocol),
1073 8 => Ok(ApplicationErrorKind::ResponseTooLarge),
1074 other => Err(ProtocolError::InvalidErrorKind(other)),
1075 }
1076}
1077
1078fn mutation_outcome_code(outcome: MutationOutcome) -> u8 {
1079 match outcome {
1080 MutationOutcome::NotApplicable => 0,
1081 MutationOutcome::NotApplied => 1,
1082 MutationOutcome::Unknown => 2,
1083 }
1084}
1085
1086fn mutation_outcome(code: u8) -> Result<MutationOutcome, ProtocolError> {
1087 match code {
1088 0 => Ok(MutationOutcome::NotApplicable),
1089 1 => Ok(MutationOutcome::NotApplied),
1090 2 => Ok(MutationOutcome::Unknown),
1091 other => Err(ProtocolError::InvalidFlag(other)),
1092 }
1093}
1094
1095struct Writer {
1096 bytes: Vec<u8>,
1097}
1098
1099impl Writer {
1100 fn new() -> Self {
1101 Self { bytes: Vec::new() }
1102 }
1103
1104 fn u8(&mut self, value: u8) {
1105 self.bytes.push(value);
1106 }
1107
1108 fn u32(&mut self, value: u32) {
1109 self.bytes.extend_from_slice(&value.to_be_bytes());
1110 }
1111
1112 fn u64(&mut self, value: u64) {
1113 self.bytes.extend_from_slice(&value.to_be_bytes());
1114 }
1115
1116 fn bytes_raw(&mut self, value: &[u8]) -> Result<(), ProtocolError> {
1117 self.u32(u32::try_from(value.len()).map_err(|_| ProtocolError::LengthOverflow)?);
1118 self.bytes.extend_from_slice(value);
1119 Ok(())
1120 }
1121
1122 fn bytes(&mut self, value: &[u8], maximum: usize) -> Result<(), ProtocolError> {
1123 encode_bytes(self, value, maximum)
1124 }
1125
1126 fn finish(self) -> Vec<u8> {
1127 self.bytes
1128 }
1129}
1130
1131struct Reader<'input> {
1132 input: &'input [u8],
1133 offset: usize,
1134}
1135
1136impl<'input> Reader<'input> {
1137 fn new(input: &'input [u8]) -> Self {
1138 Self { input, offset: 0 }
1139 }
1140
1141 fn u8(&mut self) -> Result<u8, ProtocolError> {
1142 let value = *self
1143 .input
1144 .get(self.offset)
1145 .ok_or(ProtocolError::TruncatedPayload)?;
1146 self.offset += 1;
1147 Ok(value)
1148 }
1149
1150 fn u32(&mut self) -> Result<u32, ProtocolError> {
1151 let bytes = self.take(4)?;
1152 Ok(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
1153 }
1154
1155 fn u64(&mut self) -> Result<u64, ProtocolError> {
1156 let bytes = self.take(8)?;
1157 Ok(u64::from_be_bytes([
1158 bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
1159 ]))
1160 }
1161
1162 fn bytes(&mut self, maximum: usize) -> Result<Vec<u8>, ProtocolError> {
1163 let length = usize::try_from(self.u32()?).map_err(|_| ProtocolError::LengthOverflow)?;
1164 if length > maximum {
1165 return Err(ProtocolError::PayloadTooLarge { length, maximum });
1166 }
1167 Ok(self.take(length)?.to_vec())
1168 }
1169
1170 fn optional_bytes(&mut self, maximum: usize) -> Result<Option<Vec<u8>>, ProtocolError> {
1171 match self.flag()? {
1172 false => Ok(None),
1173 true => Ok(Some(self.bytes(maximum)?)),
1174 }
1175 }
1176
1177 fn count(&mut self, maximum: usize) -> Result<usize, ProtocolError> {
1178 let count = usize::try_from(self.u32()?).map_err(|_| ProtocolError::LengthOverflow)?;
1179 if count > maximum {
1180 return Err(ProtocolError::PayloadTooLarge {
1181 length: count,
1182 maximum,
1183 });
1184 }
1185 Ok(count)
1186 }
1187
1188 fn limit(&mut self, maximum: usize) -> Result<usize, ProtocolError> {
1189 self.count(maximum)
1190 }
1191
1192 fn flag(&mut self) -> Result<bool, ProtocolError> {
1193 match self.u8()? {
1194 0 => Ok(false),
1195 1 => Ok(true),
1196 other => Err(ProtocolError::InvalidFlag(other)),
1197 }
1198 }
1199
1200 fn take(&mut self, length: usize) -> Result<&'input [u8], ProtocolError> {
1201 let end = self
1202 .offset
1203 .checked_add(length)
1204 .ok_or(ProtocolError::LengthOverflow)?;
1205 let bytes = self
1206 .input
1207 .get(self.offset..end)
1208 .ok_or(ProtocolError::TruncatedPayload)?;
1209 self.offset = end;
1210 Ok(bytes)
1211 }
1212
1213 fn finish(&self) -> Result<bool, ProtocolError> {
1214 Ok(self.offset == self.input.len())
1215 }
1216}
1217
1218#[cfg(test)]
1219mod tests {
1220 use super::*;
1221 use dodb_core::{PrimaryKey, SortKey, TenantId};
1222
1223 fn key(pk: &[u8], sk: &[u8]) -> DocumentKey {
1224 DocumentKey::new(pk.to_vec(), sk.to_vec())
1225 }
1226
1227 fn request() -> Request {
1228 Request::Transact {
1229 request: TransactionRequest::new(
1230 vec![TransactionCondition::RevisionEquals {
1231 key: key(&[0, 1], &[0xff]),
1232 expected_revision: Revision::new(7),
1233 }],
1234 vec![TransactionMutation::Put {
1235 key: key(&[0], &[1, 2]),
1236 value: vec![0, 0xff, 9],
1237 }],
1238 ),
1239 }
1240 }
1241
1242 #[test]
1243 fn every_request_round_trips() {
1244 let limits = ProtocolLimits::default();
1245 let requests = vec![
1246 Request::Get { key: key(&[], &[]) },
1247 Request::Put {
1248 key: key(&[0, 1], &[2]),
1249 value: vec![0xff, 0],
1250 },
1251 Request::Delete {
1252 key: key(&[1], &[]),
1253 },
1254 Request::Query {
1255 pk: PrimaryKey::new(vec![0, 1]),
1256 exclusive_after_sk: Some(SortKey::new(vec![0xff])),
1257 limit: 10,
1258 },
1259 Request::Scan {
1260 exclusive_after_key: Some(key(&[3], &[4])),
1261 limit: 11,
1262 },
1263 request(),
1264 ];
1265 for original in requests {
1266 let encoded = encode_request(TenantId::new(99), &original, limits).unwrap();
1267 assert_eq!(
1268 decode_request_frame(&encoded, limits).unwrap(),
1269 (TenantId::new(99), original)
1270 );
1271 }
1272 }
1273
1274 #[test]
1275 fn every_response_round_trips() {
1276 let limits = ProtocolLimits::default();
1277 let responses = vec![
1278 Response::Get(RevisionState::missing(Revision::ZERO)),
1279 Response::Get(RevisionState::present(vec![0, 0xff], Revision::new(2))),
1280 Response::Put(Revision::new(3)),
1281 Response::Delete(Revision::new(4)),
1282 Response::Query(vec![Document {
1283 key: key(&[0], &[1]),
1284 value: vec![2, 3],
1285 revision: Revision::new(5),
1286 }]),
1287 Response::Scan(Vec::new()),
1288 Response::Transact(TransactionOutcome::conditions_satisfied()),
1289 ];
1290 for original in responses {
1291 let opcode = response_opcode(&original);
1292 let encoded =
1293 encode_response(&ResponseEnvelope::Success(original.clone()), limits).unwrap();
1294 assert_eq!(
1295 decode_response_frame(&encoded, Some(opcode), limits).unwrap(),
1296 ResponseEnvelope::Success(original)
1297 );
1298 }
1299 }
1300
1301 #[test]
1302 fn conflict_and_error_round_trip() {
1303 let limits = ProtocolLimits::default();
1304 let error = ApplicationError {
1305 kind: ApplicationErrorKind::Conflict,
1306 detail: "conflict".to_owned(),
1307 mutation_outcome: MutationOutcome::NotApplied,
1308 conflict: Some(ConflictDetails {
1309 key: key(&[1], &[2]),
1310 expected: ConditionExpectation::RevisionEquals(Revision::new(8)),
1311 actual: ObservedState::missing(Revision::new(9)),
1312 }),
1313 };
1314 let encoded = encode_response(&ResponseEnvelope::Error(error.clone()), limits).unwrap();
1315 assert_eq!(
1316 decode_response_frame(&encoded, None, limits).unwrap(),
1317 ResponseEnvelope::Error(error)
1318 );
1319 }
1320
1321 #[test]
1322 fn malformed_frames_are_rejected() {
1323 let limits = ProtocolLimits::default();
1324 let valid =
1325 encode_request(TenantId::ZERO, &Request::Get { key: key(&[], &[]) }, limits).unwrap();
1326 assert!(matches!(
1327 decode_request_frame(&valid[..HEADER_SIZE - 1], limits),
1328 Err(ProtocolError::TruncatedHeader)
1329 ));
1330 let mut invalid_magic = valid.clone();
1331 invalid_magic[0] = b'X';
1332 assert!(matches!(
1333 decode_request_frame(&invalid_magic, limits),
1334 Err(ProtocolError::InvalidMagic)
1335 ));
1336 let mut invalid_version = valid.clone();
1337 invalid_version[5] = 2;
1338 assert!(matches!(
1339 decode_request_frame(&invalid_version, limits),
1340 Err(ProtocolError::UnsupportedVersion(2))
1341 ));
1342 let mut invalid_type = valid.clone();
1343 invalid_type[6] = 99;
1344 assert!(matches!(
1345 decode_request_frame(&invalid_type, limits),
1346 Err(ProtocolError::UnknownMessageType(99))
1347 ));
1348 let mut absurd_length = valid[..HEADER_SIZE].to_vec();
1349 absurd_length[7..].copy_from_slice(&u32::MAX.to_be_bytes());
1350 assert!(matches!(
1351 decode_request_frame(&absurd_length, limits),
1352 Err(ProtocolError::TruncatedPayload)
1353 ));
1354 let mut invalid_opcode = valid.clone();
1355 invalid_opcode[HEADER_SIZE + 8] = 99;
1356 assert!(matches!(
1357 decode_request_frame(&invalid_opcode, limits),
1358 Err(ProtocolError::InvalidOpcode(99))
1359 ));
1360 for reserved_opcode in [6, 7] {
1361 let mut reserved = valid.clone();
1362 reserved[HEADER_SIZE + 8] = reserved_opcode;
1363 assert!(matches!(
1364 decode_request_frame(&reserved, limits),
1365 Err(ProtocolError::InvalidOpcode(opcode)) if opcode == reserved_opcode
1366 ));
1367 }
1368 let mut trailing = valid.clone();
1369 trailing.push(1);
1370 assert!(matches!(
1371 decode_request_frame(&trailing, limits),
1372 Err(ProtocolError::TrailingBytes)
1373 ));
1374 }
1375
1376 #[test]
1377 fn response_type_mismatch_is_rejected() {
1378 let limits = ProtocolLimits::default();
1379 let encoded = encode_response(
1380 &ResponseEnvelope::Success(Response::Get(RevisionState::missing(Revision::ZERO))),
1381 limits,
1382 )
1383 .unwrap();
1384 assert!(matches!(
1385 decode_response_frame(&encoded, Some(2), limits),
1386 Err(ProtocolError::InvalidResponseType {
1387 expected: 2,
1388 actual: 1
1389 })
1390 ));
1391 }
1392
1393 #[test]
1394 fn oversized_values_are_rejected_before_allocation() {
1395 let limits = ProtocolLimits {
1396 max_value_size: 4,
1397 ..ProtocolLimits::default()
1398 };
1399 let request = Request::Put {
1400 key: key(&[1], &[2]),
1401 value: vec![1, 2, 3, 4, 5],
1402 };
1403 assert!(matches!(
1404 encode_request(TenantId::ZERO, &request, limits),
1405 Err(ProtocolError::PayloadTooLarge { .. })
1406 ));
1407 }
1408
1409 #[test]
1410 fn values_at_the_configured_limit_round_trip() {
1411 let limits = ProtocolLimits {
1412 max_value_size: 1_024,
1413 ..ProtocolLimits::default()
1414 };
1415 let request = Request::Put {
1416 key: key(&[1], &[2]),
1417 value: vec![0xff; 1_024],
1418 };
1419 let encoded = encode_request(TenantId::new(4), &request, limits).unwrap();
1420 assert_eq!(
1421 decode_request_frame(&encoded, limits).unwrap(),
1422 (TenantId::new(4), request)
1423 );
1424 }
1425
1426 #[test]
1427 fn core_error_mapping_preserves_conflict() {
1428 let conflict = TransactionConflict {
1429 key: key(&[1], &[2]),
1430 expected: ConditionExpectation::Exists,
1431 actual: ObservedState::missing(Revision::new(10)),
1432 };
1433 let mapped = ApplicationError::from_core(&Error::conflict(conflict.clone()));
1434 assert_eq!(mapped.kind, ApplicationErrorKind::Conflict);
1435 assert_eq!(mapped.conflict, Some(ConflictDetails::from(&conflict)));
1436 assert_eq!(mapped.mutation_outcome, MutationOutcome::NotApplied);
1437 assert_eq!(
1438 ApplicationError::from_core(&Error::invalid_request("bad request")).mutation_outcome,
1439 MutationOutcome::NotApplied
1440 );
1441 assert_eq!(
1442 ApplicationError::from_core(&Error::overloaded("busy")).mutation_outcome,
1443 MutationOutcome::NotApplied
1444 );
1445 }
1446
1447 #[test]
1448 fn canonical_key_limit_accounts_for_zero_byte_escaping() {
1449 let limits = ProtocolLimits::default();
1450 let boundary = Request::Get {
1451 key: key(&vec![0; 1_994], &[]),
1452 };
1453 encode_request(TenantId::ZERO, &boundary, limits).unwrap();
1454
1455 let oversized = Request::Get {
1456 key: key(&vec![0; 1_995], &[]),
1457 };
1458 assert!(matches!(
1459 encode_request(TenantId::ZERO, &oversized, limits),
1460 Err(ProtocolError::KeyTooLarge { .. })
1461 ));
1462
1463 let oversized_cursor = Request::Query {
1464 pk: PrimaryKey::new(vec![0; 1_994]),
1465 exclusive_after_sk: Some(SortKey::new(vec![0])),
1466 limit: 1,
1467 };
1468 assert!(matches!(
1469 encode_request(TenantId::ZERO, &oversized_cursor, limits),
1470 Err(ProtocolError::KeyTooLarge { .. })
1471 ));
1472 }
1473
1474 #[test]
1475 fn application_error_round_trip_preserves_unknown_mutation_outcome() {
1476 let error = ApplicationError {
1477 kind: ApplicationErrorKind::DurabilityFailure,
1478 detail: "WAL sync failed".to_owned(),
1479 mutation_outcome: MutationOutcome::Unknown,
1480 conflict: None,
1481 };
1482 let encoded = encode_response(
1483 &ResponseEnvelope::Error(error.clone()),
1484 ProtocolLimits::default(),
1485 )
1486 .unwrap();
1487 assert_eq!(
1488 decode_response_frame(&encoded, None, ProtocolLimits::default()).unwrap(),
1489 ResponseEnvelope::Error(error)
1490 );
1491 }
1492}