1use alloc::{string::String, vec::Vec};
9
10use crate::algebra::{ResourceDimension, ResourceVector, WideResourceVector};
11
12use super::codec::CodecError;
13use super::{
14 AttachAttemptToken, AttachSecret, BindingEpoch, ConnectionIncarnation, DetachAttemptToken,
15 EnrollmentToken, Generation, LeaveAttemptToken, ProtocolVersion, RecordAdmissionAttemptToken,
16 closure as c, envelope as e, response as r, tags as t,
17};
18
19const TOKEN_LEN: usize = 16;
20const SECRET_LEN: usize = 32;
21const RECOVERY_STATUS_LEN: u128 = 26;
22
23pub(super) struct TerminalizedWireDecodeAuthority(());
26
27const TERMINALIZED_WIRE_DECODE_AUTHORITY: TerminalizedWireDecodeAuthority =
28 TerminalizedWireDecodeAuthority(());
29
30pub fn encode_server_value_body(
42 value: &r::ServerValue,
43 version: ProtocolVersion,
44) -> Result<(t::ServerDiscriminant, Vec<u8>), CodecError> {
45 require_v1(version)?;
46 let discriminant = value.discriminant();
47 let mut encoder = Encoder::default();
48
49 if carries_origin(discriminant) {
50 let originating_request = value
51 .originating_request()
52 .ok_or(CodecError::InvalidValue)?;
53 encoder.put_u16(originating_request.wire_value());
54 }
55
56 encode_server_suffix(value, &mut encoder)?;
57 Ok((discriminant, encoder.bytes))
58}
59
60pub fn decode_server_value_body(
70 discriminant: t::ServerDiscriminant,
71 version: ProtocolVersion,
72 body: &[u8],
73) -> Result<(r::ServerValue, ProtocolVersion), CodecError> {
74 require_v1(version)?;
75 let mut decoder = Decoder::new(body);
76 let originating_request = if carries_origin(discriminant) {
77 let raw = decoder.take_u16()?;
78 let request = t::ClientDiscriminant::try_from(raw)
79 .map_err(|_| decode_error(t::DecodeClass::InvalidField))?;
80 if !origin_is_valid(discriminant, request) {
81 return Err(decode_error(t::DecodeClass::InvalidField));
82 }
83 Some(request)
84 } else {
85 None
86 };
87
88 let value = decode_server_suffix(discriminant, originating_request, &mut decoder)?;
89 decoder.finish()?;
90 Ok((value, version))
91}
92
93fn require_v1(version: ProtocolVersion) -> Result<(), CodecError> {
94 if version == ProtocolVersion::V1 {
95 Ok(())
96 } else {
97 Err(CodecError::UnsupportedVersion {
98 presented: version,
99 supported: ProtocolVersion::V1,
100 })
101 }
102}
103
104const fn decode_error(class: t::DecodeClass) -> CodecError {
105 CodecError::Decode { class }
106}
107
108#[derive(Default)]
109struct Encoder {
110 bytes: Vec<u8>,
111}
112
113impl Encoder {
114 fn put_u8(&mut self, value: u8) {
115 self.bytes.push(value);
116 }
117
118 fn put_u16(&mut self, value: u16) {
119 self.bytes.extend_from_slice(&value.to_be_bytes());
120 }
121
122 fn put_u32(&mut self, value: u32) {
123 self.bytes.extend_from_slice(&value.to_be_bytes());
124 }
125
126 fn put_u64(&mut self, value: u64) {
127 self.bytes.extend_from_slice(&value.to_be_bytes());
128 }
129
130 fn put_u128(&mut self, value: u128) {
131 self.bytes.extend_from_slice(&value.to_be_bytes());
132 }
133
134 fn put_fixed(&mut self, value: &[u8]) {
135 self.bytes.extend_from_slice(value);
136 }
137
138 fn put_bool(&mut self, value: bool) {
139 self.put_u8(u8::from(value));
140 }
141
142 fn put_option_u64(&mut self, value: Option<u64>) {
143 match value {
144 Some(value) => {
145 self.put_u8(1);
146 self.put_u64(value);
147 }
148 None => self.put_u8(0),
149 }
150 }
151
152 fn put_option_generation(&mut self, value: Option<Generation>) {
153 match value {
154 Some(value) => {
155 self.put_u8(1);
156 self.put_generation(value);
157 }
158 None => self.put_u8(0),
159 }
160 }
161
162 fn put_option_binding_epoch(&mut self, value: Option<BindingEpoch>) {
163 match value {
164 Some(value) => {
165 self.put_u8(1);
166 self.put_binding_epoch(value);
167 }
168 None => self.put_u8(0),
169 }
170 }
171
172 fn put_generation(&mut self, value: Generation) {
173 self.put_u64(value.get());
174 }
175
176 fn put_protocol_version(&mut self, value: ProtocolVersion) {
177 self.put_u16(value.major);
178 self.put_u16(value.minor);
179 }
180
181 fn put_binding_epoch(&mut self, value: BindingEpoch) {
182 self.put_u64(value.connection_incarnation.server_incarnation);
183 self.put_u64(value.connection_incarnation.connection_ordinal);
184 self.put_generation(value.capability_generation);
185 }
186
187 fn put_resource_vector(&mut self, value: ResourceVector) {
188 self.put_u64(value.entries);
189 self.put_u64(value.bytes);
190 }
191
192 fn put_wide_resource_vector(&mut self, value: WideResourceVector) {
193 self.put_u128(value.entries);
194 self.put_u128(value.bytes);
195 }
196
197 fn put_string(&mut self, value: &str) -> Result<(), CodecError> {
198 let length: u32 = value
199 .len()
200 .try_into()
201 .map_err(|_| CodecError::LengthOverflow)?;
202 self.put_u32(length);
203 self.put_fixed(value.as_bytes());
204 Ok(())
205 }
206}
207
208struct Decoder<'a> {
209 input: &'a [u8],
210 position: usize,
211 invalid_field: bool,
212}
213
214impl<'a> Decoder<'a> {
215 const fn new(input: &'a [u8]) -> Self {
216 Self {
217 input,
218 position: 0,
219 invalid_field: false,
220 }
221 }
222
223 const fn remaining(&self) -> usize {
224 self.input.len().saturating_sub(self.position)
225 }
226
227 fn take(&mut self, length: usize) -> Result<&'a [u8], CodecError> {
228 let end = self
229 .position
230 .checked_add(length)
231 .ok_or_else(|| decode_error(t::DecodeClass::MissingRequiredField))?;
232 let value = self
233 .input
234 .get(self.position..end)
235 .ok_or_else(|| decode_error(t::DecodeClass::MissingRequiredField))?;
236 self.position = end;
237 Ok(value)
238 }
239
240 fn take_u8(&mut self) -> Result<u8, CodecError> {
241 let bytes = self.take(1)?;
242 bytes
243 .first()
244 .copied()
245 .ok_or_else(|| decode_error(t::DecodeClass::MissingRequiredField))
246 }
247
248 fn take_u16(&mut self) -> Result<u16, CodecError> {
249 let bytes: [u8; 2] = self
250 .take(2)?
251 .try_into()
252 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
253 Ok(u16::from_be_bytes(bytes))
254 }
255
256 fn take_u32(&mut self) -> Result<u32, CodecError> {
257 let bytes: [u8; 4] = self
258 .take(4)?
259 .try_into()
260 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
261 Ok(u32::from_be_bytes(bytes))
262 }
263
264 fn take_u64(&mut self) -> Result<u64, CodecError> {
265 let bytes: [u8; 8] = self
266 .take(8)?
267 .try_into()
268 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
269 Ok(u64::from_be_bytes(bytes))
270 }
271
272 fn take_u128(&mut self) -> Result<u128, CodecError> {
273 let bytes: [u8; 16] = self
274 .take(16)?
275 .try_into()
276 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
277 Ok(u128::from_be_bytes(bytes))
278 }
279
280 fn take_fixed<const N: usize>(&mut self) -> Result<[u8; N], CodecError> {
281 self.take(N)?
282 .try_into()
283 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))
284 }
285
286 fn take_bool(&mut self) -> Result<bool, CodecError> {
287 match self.take_u8()? {
288 0 => Ok(false),
289 1 => Ok(true),
290 _ => Err(decode_error(t::DecodeClass::InvalidField)),
291 }
292 }
293
294 fn take_option_u64(&mut self) -> Result<Option<u64>, CodecError> {
295 match self.take_u8()? {
296 0 => Ok(None),
297 1 => self.take_u64().map(Some),
298 _ => Err(decode_error(t::DecodeClass::InvalidField)),
299 }
300 }
301
302 fn take_generation(&mut self) -> Result<Generation, CodecError> {
303 let raw = self.take_u64()?;
304 if let Some(value) = Generation::new(raw) {
305 Ok(value)
306 } else {
307 self.invalid_field = true;
308 generation_one()
309 }
310 }
311
312 fn take_option_generation(&mut self) -> Result<Option<Generation>, CodecError> {
313 match self.take_u8()? {
314 0 => Ok(None),
315 1 => self.take_generation().map(Some),
316 _ => Err(decode_error(t::DecodeClass::InvalidField)),
317 }
318 }
319
320 fn take_binding_epoch(&mut self) -> Result<BindingEpoch, CodecError> {
321 Ok(BindingEpoch::new(
322 ConnectionIncarnation::new(self.take_u64()?, self.take_u64()?),
323 self.take_generation()?,
324 ))
325 }
326
327 fn take_option_binding_epoch(&mut self) -> Result<Option<BindingEpoch>, CodecError> {
328 match self.take_u8()? {
329 0 => Ok(None),
330 1 => self.take_binding_epoch().map(Some),
331 _ => Err(decode_error(t::DecodeClass::InvalidField)),
332 }
333 }
334
335 fn take_protocol_version(&mut self) -> Result<ProtocolVersion, CodecError> {
336 Ok(ProtocolVersion::new(self.take_u16()?, self.take_u16()?))
337 }
338
339 fn take_resource_vector(&mut self) -> Result<ResourceVector, CodecError> {
340 Ok(ResourceVector::new(self.take_u64()?, self.take_u64()?))
341 }
342
343 fn take_wide_resource_vector(&mut self) -> Result<WideResourceVector, CodecError> {
344 Ok(WideResourceVector::new(
345 self.take_u128()?,
346 self.take_u128()?,
347 ))
348 }
349
350 fn take_string(&mut self) -> Result<String, CodecError> {
351 let length: usize = self
352 .take_u32()?
353 .try_into()
354 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
355 let bytes = self.take(length)?;
356 if let Ok(value) = core::str::from_utf8(bytes) {
357 Ok(String::from(value))
358 } else {
359 self.invalid_field = true;
360 Ok(String::new())
361 }
362 }
363
364 const fn invalidate(&mut self) {
365 self.invalid_field = true;
366 }
367
368 const fn finish(self) -> Result<(), CodecError> {
369 if self.remaining() != 0 {
370 Err(decode_error(t::DecodeClass::CanonicalEncoding))
371 } else if self.invalid_field {
372 Err(decode_error(t::DecodeClass::InvalidField))
373 } else {
374 Ok(())
375 }
376 }
377}
378
379fn generation_one() -> Result<Generation, CodecError> {
380 Generation::new(1).ok_or(CodecError::InvalidValue)
381}
382
383fn put_enrollment(value: &e::EnrollmentEnvelope, encoder: &mut Encoder) {
384 encoder.put_u64(value.conversation_id);
385 encoder.put_fixed(value.enrollment_token.as_bytes());
386}
387
388fn put_attach(value: &e::AttachEnvelope, encoder: &mut Encoder) {
389 encoder.put_u64(value.conversation_id);
390 encoder.put_u64(value.participant_id);
391 encoder.put_generation(value.capability_generation);
392 encoder.put_fixed(value.attach_attempt_token.as_bytes());
393 encoder.put_option_u64(value.accept_marker_delivery_seq);
394}
395
396fn put_detach(value: &e::DetachEnvelope, encoder: &mut Encoder) {
397 encoder.put_u64(value.conversation_id);
398 encoder.put_u64(value.participant_id);
399 encoder.put_generation(value.capability_generation);
400 encoder.put_fixed(value.detach_attempt_token.as_bytes());
401}
402
403fn put_participant_ack(value: &e::ParticipantAckEnvelope, encoder: &mut Encoder) {
404 encoder.put_u64(value.conversation_id);
405 encoder.put_u64(value.participant_id);
406 encoder.put_generation(value.capability_generation);
407 encoder.put_u64(value.through_seq);
408}
409
410fn put_leave(value: &e::LeaveEnvelope, encoder: &mut Encoder) {
411 encoder.put_u64(value.conversation_id);
412 encoder.put_u64(value.participant_id);
413 encoder.put_generation(value.capability_generation);
414 encoder.put_fixed(value.leave_attempt_token.as_bytes());
415}
416
417fn put_marker_ack(value: &e::MarkerAckEnvelope, encoder: &mut Encoder) {
418 encoder.put_u64(value.conversation_id);
419 encoder.put_u64(value.participant_id);
420 encoder.put_generation(value.capability_generation);
421 encoder.put_u64(value.marker_delivery_seq);
422}
423
424fn put_record_admission(value: &e::RecordAdmissionEnvelope, encoder: &mut Encoder) {
425 encoder.put_u64(value.conversation_id);
426 encoder.put_u64(value.participant_id);
427 encoder.put_generation(value.capability_generation);
428 encoder.put_fixed(value.record_admission_attempt_token.as_bytes());
429}
430
431fn put_response_envelope(value: &e::ResponseEnvelope, encoder: &mut Encoder) {
432 match value {
433 e::ResponseEnvelope::Enrollment(value) => put_enrollment(value, encoder),
434 e::ResponseEnvelope::CredentialAttach(value) => put_attach(value, encoder),
435 e::ResponseEnvelope::Detach(value) => put_detach(value, encoder),
436 e::ResponseEnvelope::ParticipantAck(value) => put_participant_ack(value, encoder),
437 e::ResponseEnvelope::Leave(value) => put_leave(value, encoder),
438 e::ResponseEnvelope::MarkerAck(value) => put_marker_ack(value, encoder),
439 e::ResponseEnvelope::RecordAdmission(value) => put_record_admission(value, encoder),
440 }
441}
442
443fn put_participant_reference(value: &r::ParticipantReferenceEnvelope, encoder: &mut Encoder) {
444 match value {
445 r::ParticipantReferenceEnvelope::CredentialAttach(value) => put_attach(value, encoder),
446 r::ParticipantReferenceEnvelope::Detach(value) => put_detach(value, encoder),
447 r::ParticipantReferenceEnvelope::ParticipantAck(value) => {
448 put_participant_ack(value, encoder);
449 }
450 r::ParticipantReferenceEnvelope::Leave(value) => put_leave(value, encoder),
451 r::ParticipantReferenceEnvelope::MarkerAck(value) => put_marker_ack(value, encoder),
452 r::ParticipantReferenceEnvelope::RecordAdmission(value) => {
453 put_record_admission(value, encoder);
454 }
455 }
456}
457
458fn put_binding_required(value: &r::BindingRequiredEnvelope, encoder: &mut Encoder) {
459 match value {
460 r::BindingRequiredEnvelope::Detach(value) => put_detach(value, encoder),
461 r::BindingRequiredEnvelope::ParticipantAck(value) => put_participant_ack(value, encoder),
462 r::BindingRequiredEnvelope::Leave(value) => put_leave(value, encoder),
463 r::BindingRequiredEnvelope::MarkerAck(value) => put_marker_ack(value, encoder),
464 r::BindingRequiredEnvelope::RecordAdmission(value) => put_record_admission(value, encoder),
465 }
466}
467
468fn put_order_allocating(value: &r::OrderAllocatingEnvelope, encoder: &mut Encoder) {
469 match value {
470 r::OrderAllocatingEnvelope::Enrollment(value) => put_enrollment(value, encoder),
471 r::OrderAllocatingEnvelope::CredentialAttach(value) => put_attach(value, encoder),
472 r::OrderAllocatingEnvelope::RecordAdmission(value) => put_record_admission(value, encoder),
473 }
474}
475
476fn put_closure_checked(value: &c::ClosureCheckedEnvelope, encoder: &mut Encoder) {
477 match value {
478 c::ClosureCheckedEnvelope::Enrollment(value) => put_enrollment(value, encoder),
479 c::ClosureCheckedEnvelope::CredentialAttach(value) => put_attach(value, encoder),
480 c::ClosureCheckedEnvelope::Leave(value) => put_leave(value, encoder),
481 c::ClosureCheckedEnvelope::RecordAdmission(value) => put_record_admission(value, encoder),
482 }
483}
484
485fn put_sequence_allocating(value: &r::SequenceAllocatingEnvelope, encoder: &mut Encoder) {
486 match value {
487 r::SequenceAllocatingEnvelope::Enrollment(value) => put_enrollment(value, encoder),
488 r::SequenceAllocatingEnvelope::CredentialAttach(value) => put_attach(value, encoder),
489 r::SequenceAllocatingEnvelope::RecordAdmission(value) => {
490 put_record_admission(value, encoder);
491 }
492 }
493}
494
495fn put_marker_proof(value: &r::MarkerProofRequest, encoder: &mut Encoder) {
496 match value {
497 r::MarkerProofRequest::CredentialAttach(value) => {
498 encoder.put_u64(value.conversation_id);
499 encoder.put_fixed(value.token.as_bytes());
500 encoder.put_u64(value.participant_id);
501 encoder.put_generation(value.capability_generation);
502 encoder.put_u64(value.requested_marker_delivery_seq);
503 }
504 r::MarkerProofRequest::MarkerAck(value) => {
505 encoder.put_u64(value.conversation_id);
506 encoder.put_u64(value.participant_id);
507 encoder.put_generation(value.capability_generation);
508 encoder.put_u64(value.requested_marker_delivery_seq);
509 }
510 }
511}
512
513fn put_repayment_edge(value: c::RepaymentEdge, encoder: &mut Encoder) {
514 encoder.put_u16(value.tag().wire_value());
515 match value {
516 c::RepaymentEdge::None => {}
517 c::RepaymentEdge::ObserverProjection { through_seq } => encoder.put_u64(through_seq),
518 c::RepaymentEdge::PhysicalCompaction {
519 from_floor,
520 through_seq,
521 } => {
522 encoder.put_u64(from_floor);
523 encoder.put_u64(through_seq);
524 }
525 c::RepaymentEdge::MarkerDelivery {
526 participant_id,
527 binding_epoch,
528 marker_delivery_seq,
529 } => {
530 encoder.put_u64(participant_id);
531 encoder.put_binding_epoch(binding_epoch);
532 encoder.put_u64(marker_delivery_seq);
533 }
534 c::RepaymentEdge::ParticipantCursorProgress(value) => {
535 encoder.put_u64(value.participant_id);
536 encoder.put_binding_epoch(value.binding_epoch);
537 encoder.put_u64(value.through_seq);
538 encoder.put_option_u64(value.marker_delivery_seq);
539 }
540 c::RepaymentEdge::DetachedCredentialRecovery {
541 participant_id,
542 marker_delivery_seq,
543 prior_binding_epoch,
544 } => {
545 encoder.put_u64(participant_id);
546 encoder.put_u64(marker_delivery_seq);
547 encoder.put_binding_epoch(prior_binding_epoch);
548 }
549 c::RepaymentEdge::DetachedMarkerRelease {
550 participant_id,
551 marker_delivery_seq,
552 last_dead_binding_epoch,
553 } => {
554 encoder.put_u64(participant_id);
555 encoder.put_u64(marker_delivery_seq);
556 encoder.put_binding_epoch(last_dead_binding_epoch);
557 }
558 c::RepaymentEdge::DetachedCursorRelease {
559 participant_id,
560 last_dead_binding_epoch,
561 } => {
562 encoder.put_u64(participant_id);
563 encoder.put_binding_epoch(last_dead_binding_epoch);
564 }
565 }
566}
567
568fn put_closure_snapshot(value: c::ClosureSnapshot, encoder: &mut Encoder) {
569 encoder.put_u64(value.marker_capacity_credits);
570 encoder.put_u64(value.marker_anchors);
571 encoder.put_u64(value.entry_debt);
572 encoder.put_u64(value.byte_debt);
573 put_repayment_edge(value.repayment_edge, encoder);
574 encoder.put_u64(value.edge_sequence_claims);
575 encoder.put_u64(value.edge_order_position_claims);
576 encoder.put_resource_vector(value.edge_k_remaining);
577 encoder.put_wide_resource_vector(value.k_headroom);
578 encoder.put_u64(value.episode_churn_used);
579 encoder.put_u64(value.delta_cycles);
580 encoder.put_u64(value.episode_churn_limit);
581}
582
583fn put_sequence_budget(value: super::SequenceBudget, encoder: &mut Encoder) {
584 encoder.put_u64(value.high_watermark);
585 encoder.put_u64(value.remaining);
586 encoder.put_u64(value.e);
587 encoder.put_u64(value.t);
588 encoder.put_u64(value.m);
589 encoder.put_u64(value.rs);
590 encoder.put_u64(value.rt);
591 encoder.put_u128(value.l_times_t);
592 encoder.put_u128(value.l_times_rt);
593 encoder.put_u128(value.l_other_times_e);
594}
595
596#[allow(clippy::too_many_lines)]
597fn encode_server_suffix(value: &r::ServerValue, encoder: &mut Encoder) -> Result<(), CodecError> {
598 match value {
599 r::ServerValue::ParticipantTransportRejected(value) => match &value.reason {
600 r::TransportRejectionReason::FrameTooLarge {
601 complete_frame_bytes,
602 max_frame_bytes,
603 } => {
604 encoder.put_u16(t::TransportReasonTag::FrameTooLarge.wire_value());
605 encoder.put_u64(*complete_frame_bytes);
606 encoder.put_u64(*max_frame_bytes);
607 }
608 r::TransportRejectionReason::DecodeFailed { decode_class } => {
609 encoder.put_u16(t::TransportReasonTag::DecodeFailed.wire_value());
610 encoder.put_u16(decode_class.wire_value());
611 }
612 r::TransportRejectionReason::UnsupportedVersion {
613 presented_version,
614 supported_version,
615 } => {
616 encoder.put_u16(t::TransportReasonTag::UnsupportedVersion.wire_value());
617 encoder.put_protocol_version(*presented_version);
618 encoder.put_protocol_version(*supported_version);
619 }
620 r::TransportRejectionReason::AuthenticationFailed => {
621 encoder.put_u16(t::TransportReasonTag::AuthenticationFailed.wire_value());
622 }
623 r::TransportRejectionReason::ParticipantCapabilityRequired => {
624 encoder.put_u16(t::TransportReasonTag::ParticipantCapabilityRequired.wire_value());
625 encoder.put_string(r::PARTICIPANT_CAPABILITY)?;
626 }
627 r::TransportRejectionReason::EnrollmentNotPermitted => {
628 encoder.put_u16(t::TransportReasonTag::EnrollmentNotPermitted.wire_value());
629 }
630 r::TransportRejectionReason::EnrollmentConversationOutOfScope => encoder
631 .put_u16(t::TransportReasonTag::EnrollmentConversationOutOfScope.wire_value()),
632 r::TransportRejectionReason::CredentialAttachConversationOutOfScope => encoder.put_u16(
633 t::TransportReasonTag::CredentialAttachConversationOutOfScope.wire_value(),
634 ),
635 },
636 r::ServerValue::AttemptTokenBodyConflict(value) => match value {
637 r::AttemptTokenBodyConflict::CredentialAttach {
638 token,
639 conversation_id,
640 presented_participant_id,
641 presented_generation,
642 presented_marker_delivery_seq,
643 conflict,
644 } => {
645 encoder.put_fixed(token.as_bytes());
646 encoder.put_u16(t::AttemptOperation::CredentialAttachRequest.wire_value());
647 encoder.put_u64(*conversation_id);
648 encoder.put_u64(*presented_participant_id);
649 encoder.put_generation(*presented_generation);
650 encoder.put_option_u64(*presented_marker_delivery_seq);
651 encoder.put_u16(conflict.wire_value());
652 }
653 r::AttemptTokenBodyConflict::Leave {
654 token,
655 conversation_id,
656 presented_participant_id,
657 presented_generation,
658 } => {
659 encoder.put_fixed(token.as_bytes());
660 encoder.put_u16(t::AttemptOperation::LeaveRequest.wire_value());
661 encoder.put_u64(*conversation_id);
662 encoder.put_u64(*presented_participant_id);
663 encoder.put_generation(*presented_generation);
664 encoder.put_u16(t::AttemptConflict::Generation.wire_value());
665 }
666 r::AttemptTokenBodyConflict::RecordAdmission {
667 token,
668 conversation_id,
669 presented_participant_id,
670 presented_generation,
671 } => {
672 encoder.put_fixed(token.as_bytes());
673 encoder.put_u16(t::AttemptOperation::RecordAdmission.wire_value());
674 encoder.put_u64(*conversation_id);
675 encoder.put_u64(*presented_participant_id);
676 encoder.put_generation(*presented_generation);
677 }
678 },
679 r::ServerValue::ConnectionConversationCapacityExceeded(value) => match value {
680 r::ConnectionConversationCapacityExceeded::SemanticRequest { request, limit } => {
681 put_response_envelope(request, encoder);
682 encoder.put_u64(*limit);
683 }
684 r::ConnectionConversationCapacityExceeded::ObserverRecovery {
685 conversation_id,
686 limit,
687 } => {
688 encoder.put_u64(0);
689 encoder.put_u64(*conversation_id);
690 encoder.put_u64(*limit);
691 }
692 },
693 r::ServerValue::ConnectionConversationBindingOccupied(value) => match value {
694 r::ConnectionConversationBindingOccupied::Enrollment {
695 conversation_id,
696 enrollment_token,
697 } => {
698 encoder.put_u64(*conversation_id);
699 encoder.put_fixed(enrollment_token.as_bytes());
700 encoder.put_option_u64(None);
701 }
702 r::ConnectionConversationBindingOccupied::CredentialAttach {
703 conversation_id,
704 participant_id,
705 capability_generation,
706 attach_attempt_token,
707 accept_marker_delivery_seq,
708 } => {
709 encoder.put_u64(*conversation_id);
710 encoder.put_u64(*participant_id);
711 encoder.put_generation(*capability_generation);
712 encoder.put_fixed(attach_attempt_token.as_bytes());
713 encoder.put_option_u64(*accept_marker_delivery_seq);
714 encoder.put_option_u64(Some(*participant_id));
715 }
716 },
717 r::ServerValue::ConversationOrderExhausted(value) => {
718 put_order_allocating(value.request(), encoder);
719 encoder.put_u16(value.counter().wire_value());
720 encoder.put_u64(value.high());
721 encoder.put_option_u64(value.next_value());
722 encoder.put_u128(value.order_remaining());
723 encoder.put_u128(value.reserved_claims());
724 encoder.put_u64(r::ConversationOrderExhausted::REQUIRED_MAJORS);
725 encoder.put_u128(value.resulting_order_remaining());
726 encoder.put_u128(value.resulting_reserved_claims());
727 }
728 r::ServerValue::ParticipantUnknown(value) => {
729 put_participant_reference(&value.request, encoder);
730 }
731 r::ServerValue::NoBinding(value) => put_binding_required(&value.request, encoder),
732 r::ServerValue::StaleAuthority(value) => put_stale_authority(value, encoder),
733 r::ServerValue::Retired(value) => match value {
734 r::Retired::Enrollment {
735 request,
736 participant_id,
737 retired_generation,
738 } => {
739 put_enrollment(request, encoder);
740 encoder.put_u64(*participant_id);
741 encoder.put_generation(*retired_generation);
742 }
743 r::Retired::Participant {
744 request,
745 retired_generation,
746 } => {
747 put_participant_reference(request, encoder);
748 encoder.put_generation(*retired_generation);
749 }
750 },
751 r::ServerValue::MarkerClosureCapacityExceeded(value) => {
752 put_closure_checked(&value.request, encoder);
753 let scope = match value.reason {
754 c::ClosureRefusalReason::Capacity(_) => t::ClosureScope::Capacity,
755 c::ClosureRefusalReason::RecoveryFence => t::ClosureScope::RecoveryFence,
756 c::ClosureRefusalReason::DeliveredMarkerAwaitingAck => {
757 t::ClosureScope::DeliveredMarkerAwaitingAck
758 }
759 c::ClosureRefusalReason::EpisodeChurnLimit => t::ClosureScope::EpisodeChurnLimit,
760 };
761 encoder.put_u16(scope.wire_value());
762 put_closure_snapshot(value.snapshot, encoder);
763 if let c::ClosureRefusalReason::Capacity(reason) = value.reason {
764 encoder.put_u16(t::ResourceDimensionTag::from(reason.dimension).wire_value());
765 encoder.put_u128(reason.required);
766 encoder.put_u128(reason.limit);
767 }
768 }
769 r::ServerValue::EnrollBound(value) => {
770 encoder.put_u64(value.conversation_id());
771 encoder.put_fixed(value.token().as_bytes());
772 encoder.put_u64(value.participant_id());
773 encoder.put_option_generation(value.request_generation());
774 encoder.put_generation(value.capability_generation());
775 encoder.put_fixed(value.attach_secret().as_bytes());
776 encoder.put_binding_epoch(value.origin_binding_epoch());
777 encoder.put_u64(value.persisted_cursor());
778 encoder.put_option_u64(value.accepted_marker_delivery_seq());
779 encoder.put_u128(value.receipt_expires_at());
780 encoder.put_u128(value.provenance_expires_at());
781 }
782 r::ServerValue::EnrollmentKnown(value) => {
783 encoder.put_u64(value.conversation_id);
784 encoder.put_fixed(value.token.as_bytes());
785 encoder.put_u64(value.participant_id);
786 encoder.put_generation(value.current_generation);
787 }
788 r::ServerValue::ReceiptExpired(value) => put_receipt_expired(value, encoder),
789 r::ServerValue::ReceiptCapacityExceeded(value) => {
790 put_receipt_capacity(value, encoder);
791 }
792 r::ServerValue::IdentityCapacityExceeded(value) => {
793 put_enrollment(&value.request, encoder);
794 encoder.put_u16(value.scope.wire_value());
795 encoder.put_u64(value.limit);
796 encoder.put_u64(value.occupied);
797 encoder.put_u64(r::IdentityCapacityExceeded::REQUESTED);
798 }
799 r::ServerValue::ObserverBackpressure(value) => put_observer_backpressure(value, encoder),
800 r::ServerValue::ConversationSequenceExhausted(value) => {
801 put_sequence_allocating(&value.request, encoder);
802 put_sequence_budget(value.sequence_budget, encoder);
803 }
804 r::ServerValue::AttachBound(value) => {
805 encoder.put_u64(value.conversation_id());
806 encoder.put_fixed(value.token().as_bytes());
807 encoder.put_u64(value.participant_id());
808 encoder.put_option_generation(Some(value.request_generation()));
809 encoder.put_generation(value.capability_generation());
810 encoder.put_fixed(value.attach_secret().as_bytes());
811 encoder.put_binding_epoch(value.origin_binding_epoch());
812 encoder.put_u64(value.persisted_cursor());
813 encoder.put_option_u64(value.accepted_marker_delivery_seq());
814 encoder.put_u128(value.receipt_expires_at());
815 encoder.put_u128(value.provenance_expires_at());
816 }
817 r::ServerValue::StaleOrUnknownReceipt(value) => {
818 encoder.put_u64(value.conversation_id);
819 encoder.put_fixed(value.token.as_bytes());
820 encoder.put_u64(value.participant_id);
821 encoder.put_generation(value.presented_generation);
822 encoder.put_option_u64(value.presented_marker_delivery_seq);
823 encoder.put_generation(value.current_generation);
824 }
825 r::ServerValue::MarkerNotDelivered(value) => {
826 put_marker_proof(&value.request, encoder);
827 encoder.put_u16(value.reason.wire_value());
828 encoder.put_u64(value.expected_marker_delivery_seq);
829 }
830 r::ServerValue::MarkerMismatch(value) => {
831 put_marker_proof(&value.request, encoder);
832 encoder.put_u16(value.mismatch.reason().wire_value());
833 match value.mismatch {
834 r::MarkerMismatchBody::BelowCursor { current_cursor } => {
835 encoder.put_u64(current_cursor);
836 }
837 r::MarkerMismatchBody::NoMarkerExpected => {}
838 r::MarkerMismatchBody::ExpectedDifferentMarker {
839 expected_marker_delivery_seq,
840 } => encoder.put_u64(expected_marker_delivery_seq),
841 }
842 }
843 r::ServerValue::Bound(value) | r::ServerValue::UnboundReceipt(value) => {
844 put_receipt_replay(value, encoder);
845 }
846 r::ServerValue::DetachCommitted(value) => {
847 encoder.put_u64(value.conversation_id());
848 encoder.put_u64(value.participant_id());
849 encoder.put_generation(value.capability_generation());
850 encoder.put_fixed(value.detach_attempt_token().as_bytes());
851 encoder.put_binding_epoch(value.committed_binding_epoch());
852 encoder.put_u64(value.detached_delivery_seq());
853 }
854 r::ServerValue::DetachInProgress(value) => {
855 encoder.put_u64(value.conversation_id);
856 encoder.put_u64(value.participant_id);
857 encoder.put_fixed(value.presented_token.as_bytes());
858 encoder.put_generation(value.presented_generation);
859 encoder.put_binding_epoch(value.committed_binding_epoch);
860 }
861 r::ServerValue::AckCommitted(value) => {
862 put_participant_ack(value.request(), encoder);
863 encoder.put_u64(value.current_cursor());
864 }
865 r::ServerValue::AckNoOp(value) => match value {
866 r::AckNoOp::ParticipantAck(request) => {
867 put_participant_ack(request, encoder);
868 encoder.put_u64(value.current_cursor());
869 }
870 r::AckNoOp::MarkerAck(request) => {
871 put_marker_ack(request, encoder);
872 encoder.put_u64(value.current_cursor());
873 }
874 },
875 r::ServerValue::AckGap(value) => {
876 put_participant_ack(value.request(), encoder);
877 encoder.put_u64(value.current_cursor());
878 encoder.put_u16(value.reason().wire_value());
879 }
880 r::ServerValue::AckRegression(value) => {
881 put_participant_ack(value.request(), encoder);
882 encoder.put_u64(value.current_cursor());
883 encoder.put_u16(value.reason().wire_value());
884 }
885 r::ServerValue::LeaveCommitted(value) => {
886 encoder.put_u64(value.conversation_id());
887 encoder.put_fixed(value.leave_attempt_token().as_bytes());
888 encoder.put_u64(value.participant_id());
889 encoder.put_generation(value.presented_generation());
890 encoder.put_generation(value.retired_generation());
891 encoder.put_option_binding_epoch(value.ended_binding_epoch());
892 encoder.put_option_u64(value.prior_terminal_delivery_seq());
893 encoder.put_u64(value.left_delivery_seq());
894 }
895 r::ServerValue::MarkerAckCommitted(value) => {
896 put_marker_ack(value.request(), encoder);
897 encoder.put_u64(value.current_cursor());
898 }
899 r::ServerValue::RecordCommitted(value) => {
900 put_record_admission(value.request(), encoder);
901 encoder.put_u64(value.sender_participant_id());
902 encoder.put_u64(value.delivery_seq());
903 }
904 r::ServerValue::RecordTooLarge(value) => {
905 put_record_admission(&value.request, encoder);
906 encoder.put_u16(t::ResourceDimensionTag::from(value.dimension).wire_value());
907 encoder.put_resource_vector(value.encoded_record_charge);
908 encoder.put_resource_vector(value.max_ordinary_record_charge);
909 }
910 r::ServerValue::ObserverRecoveryAccepted(value) => {
911 let count: u64 = value
912 .statuses
913 .len()
914 .try_into()
915 .map_err(|_| CodecError::LengthOverflow)?;
916 encoder.put_u64(count);
917 for status in &value.statuses {
918 encoder.put_u64(status.conversation_id);
919 encoder.put_u64(status.refused_epoch);
920 encoder.put_u64(status.current_observer_progress);
921 encoder.put_bool(status.armed);
922 encoder.put_bool(status.progressed);
923 }
924 }
925 r::ServerValue::InvalidObserverEpoch(value) => {
926 encoder.put_u64(0);
927 encoder.put_u16(value.reason().wire_value());
928 match value {
929 r::InvalidObserverEpoch::ConversationUnknown {
930 conversation_id,
931 presented_epoch,
932 } => {
933 encoder.put_u64(*conversation_id);
934 encoder.put_u64(*presented_epoch);
935 encoder.put_option_u64(None);
936 }
937 r::InvalidObserverEpoch::EpochAhead {
938 conversation_id,
939 presented_epoch,
940 current_observer_progress,
941 } => {
942 encoder.put_u64(*conversation_id);
943 encoder.put_u64(*presented_epoch);
944 encoder.put_option_u64(Some(*current_observer_progress));
945 }
946 }
947 }
948 r::ServerValue::InvalidObserverEpochList(value) => {
949 encoder.put_u64(0);
950 encoder.put_u16(value.reason().wire_value());
951 match value {
952 r::InvalidObserverEpochList::TooManyEntries {
953 presented_entries,
954 max_entries,
955 } => {
956 encoder.put_u64(*presented_entries);
957 encoder.put_u64(*max_entries);
958 }
959 r::InvalidObserverEpochList::DuplicateConversation {
960 conversation_id,
961 first_index,
962 duplicate_index,
963 } => {
964 encoder.put_u64(*conversation_id);
965 encoder.put_u64(*first_index);
966 encoder.put_u64(*duplicate_index);
967 }
968 }
969 }
970 r::ServerValue::MarkerSettlementBackpressure(value) => match value {
971 r::MarkerSettlementBackpressure::CredentialAttach {
972 conversation_id,
973 refused_epoch,
974 }
975 | r::MarkerSettlementBackpressure::Detach {
976 conversation_id,
977 refused_epoch,
978 } => {
979 encoder.put_u64(*conversation_id);
980 encoder.put_u64(*refused_epoch);
981 }
982 },
983 r::ServerValue::EnrollmentSettlementBackpressure(value) => {
984 encoder.put_u64(value.conversation_id);
985 }
986 r::ServerValue::RecordAdmissionProtocolFault(value) => {
987 put_record_admission(&value.request, encoder);
988 encoder.put_u16(value.class.tag().wire_value());
989 }
990 r::ServerValue::WriteOutcomeUnknown(value) => put_write_outcome_unknown(value, encoder),
991 }
992 Ok(())
993}
994
995fn put_write_outcome_unknown(value: &r::WriteOutcomeUnknown, encoder: &mut Encoder) {
996 match value {
997 r::WriteOutcomeUnknown::Enrollment { request } => put_enrollment(request, encoder),
998 r::WriteOutcomeUnknown::CredentialAttach { request } => put_attach(request, encoder),
999 r::WriteOutcomeUnknown::Detach { request } => put_detach(request, encoder),
1000 r::WriteOutcomeUnknown::Leave { request } => put_leave(request, encoder),
1001 r::WriteOutcomeUnknown::RecordAdmission { request } => {
1002 put_record_admission(request, encoder);
1003 }
1004 }
1005}
1006
1007fn put_stale_authority(value: &r::StaleAuthority, encoder: &mut Encoder) {
1008 match value {
1009 r::StaleAuthority::Live {
1010 request,
1011 current_generation,
1012 } => {
1013 match request {
1014 r::CommonStaleAuthorityEnvelope::CredentialAttach(value) => {
1015 put_attach(value, encoder);
1016 }
1017 r::CommonStaleAuthorityEnvelope::ParticipantAck(value) => {
1018 put_participant_ack(value, encoder);
1019 }
1020 r::CommonStaleAuthorityEnvelope::MarkerAck(value) => {
1021 put_marker_ack(value, encoder);
1022 }
1023 r::CommonStaleAuthorityEnvelope::RecordAdmission(value) => {
1024 put_record_admission(value, encoder);
1025 }
1026 }
1027 encoder.put_generation(*current_generation);
1028 }
1029 r::StaleAuthority::Detach(value) => {
1030 encoder.put_u16(value.authority_state_tag().wire_value());
1031 match value {
1032 r::DetachStaleAuthority::Live {
1033 conversation_id,
1034 participant_id,
1035 capability_generation,
1036 detach_attempt_token,
1037 current_generation,
1038 } => {
1039 encoder.put_u64(*conversation_id);
1040 encoder.put_u64(*participant_id);
1041 encoder.put_generation(*capability_generation);
1042 encoder.put_fixed(detach_attempt_token.as_bytes());
1043 encoder.put_generation(*current_generation);
1044 }
1045 r::DetachStaleAuthority::TerminalizedDetachCell(value) => {
1046 encoder.put_u64(value.conversation_id());
1047 encoder.put_u64(value.participant_id());
1048 encoder.put_generation(value.capability_generation());
1049 encoder.put_fixed(value.detach_attempt_token().as_bytes());
1050 encoder.put_generation(value.current_generation());
1051 encoder.put_binding_epoch(value.committed_binding_epoch());
1052 encoder.put_u16(value.binding_state().tag().wire_value());
1053 if let r::BindingStateView::Bound {
1054 current_binding_epoch,
1055 } = value.binding_state()
1056 {
1057 encoder.put_binding_epoch(current_binding_epoch);
1058 }
1059 }
1060 }
1061 }
1062 r::StaleAuthority::Leave(value) => {
1063 encoder.put_u16(value.authority_state_tag().wire_value());
1064 match value {
1065 r::LeaveStaleAuthority::Live {
1066 conversation_id,
1067 participant_id,
1068 presented_generation,
1069 leave_attempt_token,
1070 current_generation,
1071 } => {
1072 encoder.put_u64(*conversation_id);
1073 encoder.put_u64(*participant_id);
1074 encoder.put_generation(*presented_generation);
1075 encoder.put_fixed(leave_attempt_token.as_bytes());
1076 encoder.put_generation(*current_generation);
1077 }
1078 r::LeaveStaleAuthority::CommittedLeaveTombstone {
1079 conversation_id,
1080 participant_id,
1081 presented_generation,
1082 leave_attempt_token,
1083 retired_generation,
1084 } => {
1085 encoder.put_u64(*conversation_id);
1086 encoder.put_u64(*participant_id);
1087 encoder.put_generation(*presented_generation);
1088 encoder.put_fixed(leave_attempt_token.as_bytes());
1089 encoder.put_generation(*retired_generation);
1090 }
1091 }
1092 }
1093 }
1094}
1095
1096fn put_receipt_expired(value: &r::ReceiptExpired, encoder: &mut Encoder) {
1097 match value {
1098 r::ReceiptExpired::Enrollment {
1099 conversation_id,
1100 token,
1101 participant_id,
1102 result_generation,
1103 current_generation,
1104 reason,
1105 } => {
1106 encoder.put_u64(*conversation_id);
1107 encoder.put_fixed(token.as_bytes());
1108 encoder.put_u64(*participant_id);
1109 encoder.put_option_generation(None);
1110 encoder.put_generation(*result_generation);
1111 encoder.put_generation(*current_generation);
1112 encoder.put_u16(reason.wire_value());
1113 }
1114 r::ReceiptExpired::CredentialAttach {
1115 conversation_id,
1116 token,
1117 participant_id,
1118 presented_generation,
1119 presented_marker_delivery_seq,
1120 result_generation,
1121 current_generation,
1122 reason,
1123 } => {
1124 encoder.put_u64(*conversation_id);
1125 encoder.put_fixed(token.as_bytes());
1126 encoder.put_u64(*participant_id);
1127 encoder.put_option_generation(Some(*presented_generation));
1128 encoder.put_option_u64(*presented_marker_delivery_seq);
1129 encoder.put_generation(*result_generation);
1130 encoder.put_generation(*current_generation);
1131 encoder.put_u16(reason.wire_value());
1132 }
1133 }
1134}
1135
1136fn put_receipt_capacity(value: &r::ReceiptCapacityExceeded, encoder: &mut Encoder) {
1137 match value {
1138 r::ReceiptCapacityExceeded::Enrollment {
1139 request,
1140 scope,
1141 limit,
1142 occupied,
1143 } => {
1144 put_enrollment(request, encoder);
1145 encoder.put_u16(scope.wire_scope().wire_value());
1146 encoder.put_u64(*limit);
1147 encoder.put_u64(*occupied);
1148 encoder.put_u64(r::ReceiptCapacityExceeded::REQUESTED);
1149 }
1150 r::ReceiptCapacityExceeded::CredentialAttach {
1151 request,
1152 scope,
1153 limit,
1154 occupied,
1155 } => {
1156 put_attach(request, encoder);
1157 encoder.put_u16(scope.wire_value());
1158 encoder.put_u64(*limit);
1159 encoder.put_u64(*occupied);
1160 encoder.put_u64(r::ReceiptCapacityExceeded::REQUESTED);
1161 }
1162 }
1163}
1164
1165fn put_observer_backpressure(value: &r::ObserverBackpressure, encoder: &mut Encoder) {
1166 match value {
1167 r::ObserverBackpressure::Enrollment { request, state } => {
1168 put_enrollment(request, encoder);
1169 put_backpressure_state(*state, encoder);
1170 }
1171 r::ObserverBackpressure::CredentialAttach { request, state } => {
1172 put_attach(request, encoder);
1173 put_backpressure_state(*state, encoder);
1174 }
1175 r::ObserverBackpressure::Detach {
1176 request,
1177 committed_binding_epoch,
1178 state,
1179 } => {
1180 put_detach(request, encoder);
1181 encoder.put_binding_epoch(*committed_binding_epoch);
1182 put_backpressure_state(*state, encoder);
1183 }
1184 r::ObserverBackpressure::Leave {
1185 request,
1186 state,
1187 prior_terminal_cell_exists,
1188 } => {
1189 put_leave(request, encoder);
1190 put_backpressure_state(*state, encoder);
1191 encoder.put_bool(*prior_terminal_cell_exists);
1192 }
1193 r::ObserverBackpressure::RecordAdmission { request, state } => {
1194 put_record_admission(request, encoder);
1195 put_backpressure_state(*state, encoder);
1196 }
1197 }
1198}
1199
1200fn put_backpressure_state(value: r::ObserverBackpressureState, encoder: &mut Encoder) {
1201 encoder.put_u64(value.backpressure_epoch());
1202 encoder.put_u64(value.observer_progress());
1203}
1204
1205fn put_receipt_replay(value: &r::ReceiptReplay, encoder: &mut Encoder) {
1206 match value {
1207 r::ReceiptReplay::Enrollment(value) => {
1208 encoder.put_u64(value.conversation_id());
1209 encoder.put_fixed(value.token().as_bytes());
1210 encoder.put_u64(value.participant_id());
1211 encoder.put_option_generation(None);
1212 encoder.put_generation(value.capability_generation());
1213 encoder.put_fixed(value.attach_secret().as_bytes());
1214 encoder.put_binding_epoch(value.origin_binding_epoch());
1215 encoder.put_u64(value.persisted_cursor());
1216 encoder.put_option_u64(None);
1217 encoder.put_u128(value.receipt_expires_at());
1218 encoder.put_u128(value.provenance_expires_at());
1219 }
1220 r::ReceiptReplay::CredentialAttach(value) => {
1221 encoder.put_u64(value.conversation_id());
1222 encoder.put_fixed(value.token().as_bytes());
1223 encoder.put_u64(value.participant_id());
1224 encoder.put_option_generation(Some(value.request_generation()));
1225 encoder.put_generation(value.capability_generation());
1226 encoder.put_fixed(value.attach_secret().as_bytes());
1227 encoder.put_binding_epoch(value.origin_binding_epoch());
1228 encoder.put_u64(value.persisted_cursor());
1229 encoder.put_option_u64(value.accepted_marker_delivery_seq());
1230 encoder.put_u128(value.receipt_expires_at());
1231 encoder.put_u128(value.provenance_expires_at());
1232 }
1233 }
1234}
1235
1236const fn carries_origin(discriminant: t::ServerDiscriminant) -> bool {
1244 use t::ServerDiscriminant as D;
1245
1246 !matches!(
1247 discriminant,
1248 D::ParticipantTransportRejected
1249 | D::ObserverRecoveryAccepted
1250 | D::InvalidObserverEpoch
1251 | D::InvalidObserverEpochList
1252 | D::ObserverRecoveryConnectionCapacityExceeded
1253 )
1254}
1255
1256const fn origin_is_valid(
1257 discriminant: t::ServerDiscriminant,
1258 origin: t::ClientDiscriminant,
1259) -> bool {
1260 use t::ClientDiscriminant as O;
1261 use t::ServerDiscriminant as D;
1262
1263 match discriminant {
1264 D::AttemptTokenBodyConflict => matches!(
1265 origin,
1266 O::CredentialAttachRequest | O::LeaveRequest | O::RecordAdmission
1267 ),
1268 D::ConnectionConversationCapacityExceeded => {
1269 !matches!(origin, O::ObserverRecoveryHandshake)
1270 }
1271 D::ConnectionConversationBindingOccupied => {
1272 matches!(origin, O::EnrollmentRequest | O::CredentialAttachRequest)
1273 }
1274 D::ConversationOrderExhausted => matches!(
1275 origin,
1276 O::EnrollmentRequest | O::CredentialAttachRequest | O::RecordAdmission
1277 ),
1278 D::ParticipantUnknown => {
1279 !matches!(origin, O::EnrollmentRequest | O::ObserverRecoveryHandshake)
1280 }
1281 D::NoBinding => !matches!(
1282 origin,
1283 O::EnrollmentRequest | O::CredentialAttachRequest | O::ObserverRecoveryHandshake
1284 ),
1285 D::StaleAuthority => !matches!(origin, O::EnrollmentRequest | O::ObserverRecoveryHandshake),
1286 D::Retired => !matches!(origin, O::ObserverRecoveryHandshake),
1287 D::MarkerClosureCapacityExceeded => matches!(
1288 origin,
1289 O::EnrollmentRequest
1290 | O::CredentialAttachRequest
1291 | O::LeaveRequest
1292 | O::RecordAdmission
1293 ),
1294 D::EnrollBound | D::EnrollmentKnown | D::IdentityCapacityExceeded => {
1295 matches!(origin, O::EnrollmentRequest)
1296 }
1297 D::ReceiptExpired | D::ReceiptCapacityExceeded | D::Bound | D::UnboundReceipt => {
1298 matches!(origin, O::EnrollmentRequest | O::CredentialAttachRequest)
1299 }
1300 D::ObserverBackpressure | D::WriteOutcomeUnknown => matches!(
1301 origin,
1302 O::EnrollmentRequest
1303 | O::CredentialAttachRequest
1304 | O::DetachRequest
1305 | O::LeaveRequest
1306 | O::RecordAdmission
1307 ),
1308 D::ConversationSequenceExhausted => matches!(
1309 origin,
1310 O::EnrollmentRequest | O::CredentialAttachRequest | O::RecordAdmission
1311 ),
1312 D::AttachBound | D::StaleOrUnknownReceipt => {
1313 matches!(origin, O::CredentialAttachRequest)
1314 }
1315 D::MarkerNotDelivered | D::MarkerMismatch => {
1316 matches!(origin, O::CredentialAttachRequest | O::MarkerAck)
1317 }
1318 D::DetachCommitted | D::DetachInProgress => matches!(origin, O::DetachRequest),
1319 D::AckCommitted | D::AckGap | D::AckRegression => matches!(origin, O::ParticipantAck),
1320 D::AckNoOp => matches!(origin, O::ParticipantAck | O::MarkerAck),
1321 D::LeaveCommitted => matches!(origin, O::LeaveRequest),
1322 D::MarkerAckCommitted => matches!(origin, O::MarkerAck),
1323 D::RecordCommitted | D::RecordTooLarge | D::RecordAdmissionProtocolFault => {
1324 matches!(origin, O::RecordAdmission)
1325 }
1326 D::MarkerSettlementBackpressure => {
1327 matches!(origin, O::CredentialAttachRequest | O::DetachRequest)
1328 }
1329 D::EnrollmentSettlementBackpressure => matches!(origin, O::EnrollmentRequest),
1330 D::ParticipantTransportRejected
1331 | D::ObserverRecoveryAccepted
1332 | D::InvalidObserverEpoch
1333 | D::InvalidObserverEpochList
1334 | D::ObserverRecoveryConnectionCapacityExceeded => false,
1335 }
1336}
1337
1338fn take_enrollment(decoder: &mut Decoder<'_>) -> Result<e::EnrollmentEnvelope, CodecError> {
1342 Ok(e::EnrollmentEnvelope {
1343 conversation_id: decoder.take_u64()?,
1344 enrollment_token: EnrollmentToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1345 })
1346}
1347
1348fn take_attach(decoder: &mut Decoder<'_>) -> Result<e::AttachEnvelope, CodecError> {
1349 Ok(e::AttachEnvelope {
1350 conversation_id: decoder.take_u64()?,
1351 participant_id: decoder.take_u64()?,
1352 capability_generation: decoder.take_generation()?,
1353 attach_attempt_token: AttachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1354 accept_marker_delivery_seq: decoder.take_option_u64()?,
1355 })
1356}
1357
1358fn take_detach(decoder: &mut Decoder<'_>) -> Result<e::DetachEnvelope, CodecError> {
1359 Ok(e::DetachEnvelope {
1360 conversation_id: decoder.take_u64()?,
1361 participant_id: decoder.take_u64()?,
1362 capability_generation: decoder.take_generation()?,
1363 detach_attempt_token: DetachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1364 })
1365}
1366
1367fn take_participant_ack(
1368 decoder: &mut Decoder<'_>,
1369) -> Result<e::ParticipantAckEnvelope, CodecError> {
1370 Ok(e::ParticipantAckEnvelope {
1371 conversation_id: decoder.take_u64()?,
1372 participant_id: decoder.take_u64()?,
1373 capability_generation: decoder.take_generation()?,
1374 through_seq: decoder.take_u64()?,
1375 })
1376}
1377
1378fn take_leave(decoder: &mut Decoder<'_>) -> Result<e::LeaveEnvelope, CodecError> {
1379 Ok(e::LeaveEnvelope {
1380 conversation_id: decoder.take_u64()?,
1381 participant_id: decoder.take_u64()?,
1382 capability_generation: decoder.take_generation()?,
1383 leave_attempt_token: LeaveAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1384 })
1385}
1386
1387fn take_marker_ack(decoder: &mut Decoder<'_>) -> Result<e::MarkerAckEnvelope, CodecError> {
1388 Ok(e::MarkerAckEnvelope {
1389 conversation_id: decoder.take_u64()?,
1390 participant_id: decoder.take_u64()?,
1391 capability_generation: decoder.take_generation()?,
1392 marker_delivery_seq: decoder.take_u64()?,
1393 })
1394}
1395
1396fn take_record_admission(
1397 decoder: &mut Decoder<'_>,
1398) -> Result<e::RecordAdmissionEnvelope, CodecError> {
1399 Ok(e::RecordAdmissionEnvelope {
1400 conversation_id: decoder.take_u64()?,
1401 participant_id: decoder.take_u64()?,
1402 capability_generation: decoder.take_generation()?,
1403 record_admission_attempt_token: RecordAdmissionAttemptToken::new(
1404 decoder.take_fixed::<TOKEN_LEN>()?,
1405 ),
1406 })
1407}
1408
1409fn take_response_envelope(
1410 origin: t::ClientDiscriminant,
1411 decoder: &mut Decoder<'_>,
1412) -> Result<e::ResponseEnvelope, CodecError> {
1413 match origin {
1414 t::ClientDiscriminant::EnrollmentRequest => {
1415 take_enrollment(decoder).map(e::ResponseEnvelope::Enrollment)
1416 }
1417 t::ClientDiscriminant::CredentialAttachRequest => {
1418 take_attach(decoder).map(e::ResponseEnvelope::CredentialAttach)
1419 }
1420 t::ClientDiscriminant::DetachRequest => {
1421 take_detach(decoder).map(e::ResponseEnvelope::Detach)
1422 }
1423 t::ClientDiscriminant::ParticipantAck => {
1424 take_participant_ack(decoder).map(e::ResponseEnvelope::ParticipantAck)
1425 }
1426 t::ClientDiscriminant::LeaveRequest => take_leave(decoder).map(e::ResponseEnvelope::Leave),
1427 t::ClientDiscriminant::MarkerAck => {
1428 take_marker_ack(decoder).map(e::ResponseEnvelope::MarkerAck)
1429 }
1430 t::ClientDiscriminant::RecordAdmission => {
1431 take_record_admission(decoder).map(e::ResponseEnvelope::RecordAdmission)
1432 }
1433 t::ClientDiscriminant::ObserverRecoveryHandshake => {
1434 Err(decode_error(t::DecodeClass::InvalidField))
1435 }
1436 }
1437}
1438
1439fn take_participant_reference(
1440 origin: t::ClientDiscriminant,
1441 decoder: &mut Decoder<'_>,
1442) -> Result<r::ParticipantReferenceEnvelope, CodecError> {
1443 match origin {
1444 t::ClientDiscriminant::CredentialAttachRequest => {
1445 take_attach(decoder).map(r::ParticipantReferenceEnvelope::CredentialAttach)
1446 }
1447 t::ClientDiscriminant::DetachRequest => {
1448 take_detach(decoder).map(r::ParticipantReferenceEnvelope::Detach)
1449 }
1450 t::ClientDiscriminant::ParticipantAck => {
1451 take_participant_ack(decoder).map(r::ParticipantReferenceEnvelope::ParticipantAck)
1452 }
1453 t::ClientDiscriminant::LeaveRequest => {
1454 take_leave(decoder).map(r::ParticipantReferenceEnvelope::Leave)
1455 }
1456 t::ClientDiscriminant::MarkerAck => {
1457 take_marker_ack(decoder).map(r::ParticipantReferenceEnvelope::MarkerAck)
1458 }
1459 t::ClientDiscriminant::RecordAdmission => {
1460 take_record_admission(decoder).map(r::ParticipantReferenceEnvelope::RecordAdmission)
1461 }
1462 t::ClientDiscriminant::EnrollmentRequest
1463 | t::ClientDiscriminant::ObserverRecoveryHandshake => {
1464 Err(decode_error(t::DecodeClass::InvalidField))
1465 }
1466 }
1467}
1468
1469fn take_binding_required(
1470 origin: t::ClientDiscriminant,
1471 decoder: &mut Decoder<'_>,
1472) -> Result<r::BindingRequiredEnvelope, CodecError> {
1473 match origin {
1474 t::ClientDiscriminant::DetachRequest => {
1475 take_detach(decoder).map(r::BindingRequiredEnvelope::Detach)
1476 }
1477 t::ClientDiscriminant::ParticipantAck => {
1478 take_participant_ack(decoder).map(r::BindingRequiredEnvelope::ParticipantAck)
1479 }
1480 t::ClientDiscriminant::LeaveRequest => {
1481 take_leave(decoder).map(r::BindingRequiredEnvelope::Leave)
1482 }
1483 t::ClientDiscriminant::MarkerAck => {
1484 take_marker_ack(decoder).map(r::BindingRequiredEnvelope::MarkerAck)
1485 }
1486 t::ClientDiscriminant::RecordAdmission => {
1487 take_record_admission(decoder).map(r::BindingRequiredEnvelope::RecordAdmission)
1488 }
1489 t::ClientDiscriminant::EnrollmentRequest
1490 | t::ClientDiscriminant::CredentialAttachRequest
1491 | t::ClientDiscriminant::ObserverRecoveryHandshake => {
1492 Err(decode_error(t::DecodeClass::InvalidField))
1493 }
1494 }
1495}
1496
1497fn take_order_allocating(
1498 origin: t::ClientDiscriminant,
1499 decoder: &mut Decoder<'_>,
1500) -> Result<r::OrderAllocatingEnvelope, CodecError> {
1501 match origin {
1502 t::ClientDiscriminant::EnrollmentRequest => {
1503 take_enrollment(decoder).map(r::OrderAllocatingEnvelope::Enrollment)
1504 }
1505 t::ClientDiscriminant::CredentialAttachRequest => {
1506 take_attach(decoder).map(r::OrderAllocatingEnvelope::CredentialAttach)
1507 }
1508 t::ClientDiscriminant::RecordAdmission => {
1509 take_record_admission(decoder).map(r::OrderAllocatingEnvelope::RecordAdmission)
1510 }
1511 _ => Err(decode_error(t::DecodeClass::InvalidField)),
1512 }
1513}
1514
1515fn take_closure_checked(
1516 origin: t::ClientDiscriminant,
1517 decoder: &mut Decoder<'_>,
1518) -> Result<c::ClosureCheckedEnvelope, CodecError> {
1519 match origin {
1520 t::ClientDiscriminant::EnrollmentRequest => {
1521 take_enrollment(decoder).map(c::ClosureCheckedEnvelope::Enrollment)
1522 }
1523 t::ClientDiscriminant::CredentialAttachRequest => {
1524 take_attach(decoder).map(c::ClosureCheckedEnvelope::CredentialAttach)
1525 }
1526 t::ClientDiscriminant::LeaveRequest => {
1527 take_leave(decoder).map(c::ClosureCheckedEnvelope::Leave)
1528 }
1529 t::ClientDiscriminant::RecordAdmission => {
1530 take_record_admission(decoder).map(c::ClosureCheckedEnvelope::RecordAdmission)
1531 }
1532 _ => Err(decode_error(t::DecodeClass::InvalidField)),
1533 }
1534}
1535
1536fn take_sequence_allocating(
1537 origin: t::ClientDiscriminant,
1538 decoder: &mut Decoder<'_>,
1539) -> Result<r::SequenceAllocatingEnvelope, CodecError> {
1540 match origin {
1541 t::ClientDiscriminant::EnrollmentRequest => {
1542 take_enrollment(decoder).map(r::SequenceAllocatingEnvelope::Enrollment)
1543 }
1544 t::ClientDiscriminant::CredentialAttachRequest => {
1545 take_attach(decoder).map(r::SequenceAllocatingEnvelope::CredentialAttach)
1546 }
1547 t::ClientDiscriminant::RecordAdmission => {
1548 take_record_admission(decoder).map(r::SequenceAllocatingEnvelope::RecordAdmission)
1549 }
1550 _ => Err(decode_error(t::DecodeClass::InvalidField)),
1551 }
1552}
1553
1554fn take_marker_proof(
1555 origin: t::ClientDiscriminant,
1556 decoder: &mut Decoder<'_>,
1557) -> Result<r::MarkerProofRequest, CodecError> {
1558 match origin {
1559 t::ClientDiscriminant::CredentialAttachRequest => Ok(
1560 r::MarkerProofRequest::CredentialAttach(r::AttachMarkerProof {
1561 conversation_id: decoder.take_u64()?,
1562 token: AttachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1563 participant_id: decoder.take_u64()?,
1564 capability_generation: decoder.take_generation()?,
1565 requested_marker_delivery_seq: decoder.take_u64()?,
1566 }),
1567 ),
1568 t::ClientDiscriminant::MarkerAck => {
1569 Ok(r::MarkerProofRequest::MarkerAck(r::MarkerAckProof {
1570 conversation_id: decoder.take_u64()?,
1571 participant_id: decoder.take_u64()?,
1572 capability_generation: decoder.take_generation()?,
1573 requested_marker_delivery_seq: decoder.take_u64()?,
1574 }))
1575 }
1576 _ => Err(decode_error(t::DecodeClass::InvalidField)),
1577 }
1578}
1579
1580fn take_repayment_edge(decoder: &mut Decoder<'_>) -> Result<c::RepaymentEdge, CodecError> {
1581 let tag = t::RepaymentEdgeTag::try_from(decoder.take_u16()?)
1582 .map_err(|_| decode_error(t::DecodeClass::InvalidField))?;
1583 match tag {
1584 t::RepaymentEdgeTag::None => Ok(c::RepaymentEdge::None),
1585 t::RepaymentEdgeTag::ObserverProjection => Ok(c::RepaymentEdge::ObserverProjection {
1586 through_seq: decoder.take_u64()?,
1587 }),
1588 t::RepaymentEdgeTag::PhysicalCompaction => Ok(c::RepaymentEdge::PhysicalCompaction {
1589 from_floor: decoder.take_u64()?,
1590 through_seq: decoder.take_u64()?,
1591 }),
1592 t::RepaymentEdgeTag::MarkerDelivery => Ok(c::RepaymentEdge::MarkerDelivery {
1593 participant_id: decoder.take_u64()?,
1594 binding_epoch: decoder.take_binding_epoch()?,
1595 marker_delivery_seq: decoder.take_u64()?,
1596 }),
1597 t::RepaymentEdgeTag::ParticipantCursorProgress => Ok(
1598 c::RepaymentEdge::ParticipantCursorProgress(c::ParticipantCursorProgressEdge {
1599 participant_id: decoder.take_u64()?,
1600 binding_epoch: decoder.take_binding_epoch()?,
1601 through_seq: decoder.take_u64()?,
1602 marker_delivery_seq: decoder.take_option_u64()?,
1603 }),
1604 ),
1605 t::RepaymentEdgeTag::DetachedCredentialRecovery => {
1606 Ok(c::RepaymentEdge::DetachedCredentialRecovery {
1607 participant_id: decoder.take_u64()?,
1608 marker_delivery_seq: decoder.take_u64()?,
1609 prior_binding_epoch: decoder.take_binding_epoch()?,
1610 })
1611 }
1612 t::RepaymentEdgeTag::DetachedMarkerRelease => Ok(c::RepaymentEdge::DetachedMarkerRelease {
1613 participant_id: decoder.take_u64()?,
1614 marker_delivery_seq: decoder.take_u64()?,
1615 last_dead_binding_epoch: decoder.take_binding_epoch()?,
1616 }),
1617 t::RepaymentEdgeTag::DetachedCursorRelease => Ok(c::RepaymentEdge::DetachedCursorRelease {
1618 participant_id: decoder.take_u64()?,
1619 last_dead_binding_epoch: decoder.take_binding_epoch()?,
1620 }),
1621 }
1622}
1623
1624fn take_closure_snapshot(decoder: &mut Decoder<'_>) -> Result<c::ClosureSnapshot, CodecError> {
1625 Ok(c::ClosureSnapshot {
1626 marker_capacity_credits: decoder.take_u64()?,
1627 marker_anchors: decoder.take_u64()?,
1628 entry_debt: decoder.take_u64()?,
1629 byte_debt: decoder.take_u64()?,
1630 repayment_edge: take_repayment_edge(decoder)?,
1631 edge_sequence_claims: decoder.take_u64()?,
1632 edge_order_position_claims: decoder.take_u64()?,
1633 edge_k_remaining: decoder.take_resource_vector()?,
1634 k_headroom: decoder.take_wide_resource_vector()?,
1635 episode_churn_used: decoder.take_u64()?,
1636 delta_cycles: decoder.take_u64()?,
1637 episode_churn_limit: decoder.take_u64()?,
1638 })
1639}
1640
1641fn take_sequence_budget(decoder: &mut Decoder<'_>) -> Result<super::SequenceBudget, CodecError> {
1642 Ok(super::SequenceBudget {
1643 high_watermark: decoder.take_u64()?,
1644 remaining: decoder.take_u64()?,
1645 e: decoder.take_u64()?,
1646 t: decoder.take_u64()?,
1647 m: decoder.take_u64()?,
1648 rs: decoder.take_u64()?,
1649 rt: decoder.take_u64()?,
1650 l_times_t: decoder.take_u128()?,
1651 l_times_rt: decoder.take_u128()?,
1652 l_other_times_e: decoder.take_u128()?,
1653 })
1654}
1655
1656fn required_origin(
1657 origin: Option<t::ClientDiscriminant>,
1658) -> Result<t::ClientDiscriminant, CodecError> {
1659 origin.ok_or_else(|| decode_error(t::DecodeClass::InvalidField))
1660}
1661
1662#[allow(clippy::too_many_lines)]
1663fn decode_server_suffix(
1664 discriminant: t::ServerDiscriminant,
1665 origin: Option<t::ClientDiscriminant>,
1666 decoder: &mut Decoder<'_>,
1667) -> Result<r::ServerValue, CodecError> {
1668 use t::ServerDiscriminant as D;
1669
1670 match discriminant {
1671 D::ParticipantTransportRejected => {
1672 let reason = match take_tag::<t::TransportReasonTag>(decoder)? {
1673 t::TransportReasonTag::FrameTooLarge => {
1674 r::TransportRejectionReason::FrameTooLarge {
1675 complete_frame_bytes: decoder.take_u64()?,
1676 max_frame_bytes: decoder.take_u64()?,
1677 }
1678 }
1679 t::TransportReasonTag::DecodeFailed => {
1680 let decode_class = take_tag::<t::DecodeClass>(decoder)?;
1681 r::TransportRejectionReason::DecodeFailed { decode_class }
1682 }
1683 t::TransportReasonTag::UnsupportedVersion => {
1684 r::TransportRejectionReason::UnsupportedVersion {
1685 presented_version: decoder.take_protocol_version()?,
1686 supported_version: decoder.take_protocol_version()?,
1687 }
1688 }
1689 t::TransportReasonTag::AuthenticationFailed => {
1690 r::TransportRejectionReason::AuthenticationFailed
1691 }
1692 t::TransportReasonTag::ParticipantCapabilityRequired => {
1693 let capability = decoder.take_string()?;
1694 if capability != r::PARTICIPANT_CAPABILITY {
1695 decoder.invalidate();
1696 }
1697 r::TransportRejectionReason::ParticipantCapabilityRequired
1698 }
1699 t::TransportReasonTag::EnrollmentNotPermitted => {
1700 r::TransportRejectionReason::EnrollmentNotPermitted
1701 }
1702 t::TransportReasonTag::EnrollmentConversationOutOfScope => {
1703 r::TransportRejectionReason::EnrollmentConversationOutOfScope
1704 }
1705 t::TransportReasonTag::CredentialAttachConversationOutOfScope => {
1706 r::TransportRejectionReason::CredentialAttachConversationOutOfScope
1707 }
1708 };
1709 Ok(r::ServerValue::ParticipantTransportRejected(
1710 r::ParticipantTransportRejected { reason },
1711 ))
1712 }
1713 D::AttemptTokenBodyConflict => {
1714 let origin = required_origin(origin)?;
1715 let token = decoder.take_fixed::<TOKEN_LEN>()?;
1716 let operation = take_tag::<t::AttemptOperation>(decoder)?;
1717 match origin {
1718 t::ClientDiscriminant::CredentialAttachRequest => {
1719 if operation != t::AttemptOperation::CredentialAttachRequest {
1720 return Err(decode_error(t::DecodeClass::InvalidField));
1721 }
1722 Ok(r::ServerValue::AttemptTokenBodyConflict(
1723 r::AttemptTokenBodyConflict::CredentialAttach {
1724 token: AttachAttemptToken::new(token),
1725 conversation_id: decoder.take_u64()?,
1726 presented_participant_id: decoder.take_u64()?,
1727 presented_generation: decoder.take_generation()?,
1728 presented_marker_delivery_seq: decoder.take_option_u64()?,
1729 conflict: take_tag::<t::AttemptConflict>(decoder)?,
1730 },
1731 ))
1732 }
1733 t::ClientDiscriminant::LeaveRequest => {
1734 if operation != t::AttemptOperation::LeaveRequest {
1735 return Err(decode_error(t::DecodeClass::InvalidField));
1736 }
1737 let value = r::AttemptTokenBodyConflict::Leave {
1738 token: LeaveAttemptToken::new(token),
1739 conversation_id: decoder.take_u64()?,
1740 presented_participant_id: decoder.take_u64()?,
1741 presented_generation: decoder.take_generation()?,
1742 };
1743 if take_tag::<t::AttemptConflict>(decoder)? != t::AttemptConflict::Generation {
1744 decoder.invalidate();
1745 }
1746 Ok(r::ServerValue::AttemptTokenBodyConflict(value))
1747 }
1748 t::ClientDiscriminant::RecordAdmission => {
1749 if operation != t::AttemptOperation::RecordAdmission {
1750 return Err(decode_error(t::DecodeClass::InvalidField));
1751 }
1752 Ok(r::ServerValue::AttemptTokenBodyConflict(
1753 r::AttemptTokenBodyConflict::RecordAdmission {
1754 token: RecordAdmissionAttemptToken::new(token),
1755 conversation_id: decoder.take_u64()?,
1756 presented_participant_id: decoder.take_u64()?,
1757 presented_generation: decoder.take_generation()?,
1758 },
1759 ))
1760 }
1761 _ => Err(decode_error(t::DecodeClass::InvalidField)),
1762 }
1763 }
1764 D::ConnectionConversationCapacityExceeded => {
1765 let origin = required_origin(origin)?;
1766 Ok(r::ServerValue::ConnectionConversationCapacityExceeded(
1767 r::ConnectionConversationCapacityExceeded::SemanticRequest {
1768 request: take_response_envelope(origin, decoder)?,
1769 limit: decoder.take_u64()?,
1770 },
1771 ))
1772 }
1773 D::ConnectionConversationBindingOccupied => {
1774 let origin = required_origin(origin)?;
1775 let value = match origin {
1776 t::ClientDiscriminant::EnrollmentRequest => {
1777 let conversation_id = decoder.take_u64()?;
1778 let enrollment_token = EnrollmentToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
1779 if decoder.take_option_u64()?.is_some() {
1780 decoder.invalidate();
1781 }
1782 r::ConnectionConversationBindingOccupied::Enrollment {
1783 conversation_id,
1784 enrollment_token,
1785 }
1786 }
1787 t::ClientDiscriminant::CredentialAttachRequest => {
1788 let conversation_id = decoder.take_u64()?;
1789 let participant_id = decoder.take_u64()?;
1790 let capability_generation = decoder.take_generation()?;
1791 let attach_attempt_token =
1792 AttachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
1793 let accept_marker_delivery_seq = decoder.take_option_u64()?;
1794 if decoder.take_option_u64()? != Some(participant_id) {
1795 decoder.invalidate();
1796 }
1797 r::ConnectionConversationBindingOccupied::CredentialAttach {
1798 conversation_id,
1799 participant_id,
1800 capability_generation,
1801 attach_attempt_token,
1802 accept_marker_delivery_seq,
1803 }
1804 }
1805 _ => return Err(decode_error(t::DecodeClass::InvalidField)),
1806 };
1807 Ok(r::ServerValue::ConnectionConversationBindingOccupied(value))
1808 }
1809 D::ConversationOrderExhausted => {
1810 let origin = required_origin(origin)?;
1811 let request = take_order_allocating(origin, decoder)?;
1812 let _counter = take_tag::<t::Counter>(decoder)?;
1813 let high = decoder.take_u64()?;
1814 let next_value = decoder.take_option_u64()?;
1815 let order_remaining = decoder.take_u128()?;
1816 let reserved_claims = decoder.take_u128()?;
1817 if decoder.take_u64()? != r::ConversationOrderExhausted::REQUIRED_MAJORS {
1818 decoder.invalidate();
1819 }
1820 let resulting_order_remaining = decoder.take_u128()?;
1821 let resulting_reserved_claims = decoder.take_u128()?;
1822 let value = r::ConversationOrderExhausted::new(
1823 request,
1824 high,
1825 order_remaining,
1826 reserved_claims,
1827 resulting_order_remaining,
1828 resulting_reserved_claims,
1829 );
1830 if next_value != value.next_value() {
1831 decoder.invalidate();
1832 }
1833 Ok(r::ServerValue::ConversationOrderExhausted(
1834 alloc::boxed::Box::new(value),
1835 ))
1836 }
1837 D::ParticipantUnknown => {
1838 let origin = required_origin(origin)?;
1839 Ok(r::ServerValue::ParticipantUnknown(r::ParticipantUnknown {
1840 request: take_participant_reference(origin, decoder)?,
1841 }))
1842 }
1843 D::NoBinding => {
1844 let origin = required_origin(origin)?;
1845 Ok(r::ServerValue::NoBinding(r::NoBinding {
1846 request: take_binding_required(origin, decoder)?,
1847 }))
1848 }
1849 D::StaleAuthority => take_stale_authority(required_origin(origin)?, decoder)
1850 .map(r::ServerValue::StaleAuthority),
1851 D::Retired => {
1852 let origin = required_origin(origin)?;
1853 let value = if origin == t::ClientDiscriminant::EnrollmentRequest {
1854 r::Retired::Enrollment {
1855 request: take_enrollment(decoder)?,
1856 participant_id: decoder.take_u64()?,
1857 retired_generation: decoder.take_generation()?,
1858 }
1859 } else {
1860 r::Retired::Participant {
1861 request: take_participant_reference(origin, decoder)?,
1862 retired_generation: decoder.take_generation()?,
1863 }
1864 };
1865 Ok(r::ServerValue::Retired(value))
1866 }
1867 D::MarkerClosureCapacityExceeded => {
1868 let origin = required_origin(origin)?;
1869 let request = take_closure_checked(origin, decoder)?;
1870 let scope = take_tag::<t::ClosureScope>(decoder)?;
1871 let snapshot = take_closure_snapshot(decoder)?;
1872 let reason = match scope {
1873 t::ClosureScope::Capacity => {
1874 let dimension =
1875 ResourceDimension::from(take_tag::<t::ResourceDimensionTag>(decoder)?);
1876 c::ClosureRefusalReason::Capacity(c::ClosureCapacityReason {
1877 dimension,
1878 required: decoder.take_u128()?,
1879 limit: decoder.take_u128()?,
1880 })
1881 }
1882 t::ClosureScope::RecoveryFence => c::ClosureRefusalReason::RecoveryFence,
1883 t::ClosureScope::DeliveredMarkerAwaitingAck => {
1884 c::ClosureRefusalReason::DeliveredMarkerAwaitingAck
1885 }
1886 t::ClosureScope::EpisodeChurnLimit => c::ClosureRefusalReason::EpisodeChurnLimit,
1887 };
1888 Ok(r::ServerValue::MarkerClosureCapacityExceeded(
1889 alloc::boxed::Box::new(c::MarkerClosureCapacityExceeded {
1890 request,
1891 snapshot,
1892 reason,
1893 }),
1894 ))
1895 }
1896 D::EnrollBound => take_enroll_bound(decoder).map(r::ServerValue::EnrollBound),
1897 D::EnrollmentKnown => Ok(r::ServerValue::EnrollmentKnown(r::EnrollmentKnown {
1898 conversation_id: decoder.take_u64()?,
1899 token: EnrollmentToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1900 participant_id: decoder.take_u64()?,
1901 current_generation: decoder.take_generation()?,
1902 })),
1903 D::ReceiptExpired => take_receipt_expired(required_origin(origin)?, decoder)
1904 .map(r::ServerValue::ReceiptExpired),
1905 D::ReceiptCapacityExceeded => take_receipt_capacity(required_origin(origin)?, decoder)
1906 .map(r::ServerValue::ReceiptCapacityExceeded),
1907 D::IdentityCapacityExceeded => {
1908 let request = take_enrollment(decoder)?;
1909 let scope = take_tag::<t::IdentityCapacityScope>(decoder)?;
1910 let limit = decoder.take_u64()?;
1911 let occupied = decoder.take_u64()?;
1912 if decoder.take_u64()? != r::IdentityCapacityExceeded::REQUESTED {
1913 decoder.invalidate();
1914 }
1915 Ok(r::ServerValue::IdentityCapacityExceeded(
1916 r::IdentityCapacityExceeded {
1917 request,
1918 scope,
1919 limit,
1920 occupied,
1921 },
1922 ))
1923 }
1924 D::ObserverBackpressure => take_observer_backpressure(required_origin(origin)?, decoder)
1925 .map(r::ServerValue::ObserverBackpressure),
1926 D::ConversationSequenceExhausted => {
1927 let request = take_sequence_allocating(required_origin(origin)?, decoder)?;
1928 Ok(r::ServerValue::ConversationSequenceExhausted(
1929 alloc::boxed::Box::new(r::ConversationSequenceExhausted {
1930 request,
1931 sequence_budget: take_sequence_budget(decoder)?,
1932 }),
1933 ))
1934 }
1935 D::AttachBound => take_attach_bound(decoder).map(r::ServerValue::AttachBound),
1936 D::StaleOrUnknownReceipt => Ok(r::ServerValue::StaleOrUnknownReceipt(
1937 r::StaleOrUnknownReceipt {
1938 conversation_id: decoder.take_u64()?,
1939 token: AttachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
1940 participant_id: decoder.take_u64()?,
1941 presented_generation: decoder.take_generation()?,
1942 presented_marker_delivery_seq: decoder.take_option_u64()?,
1943 current_generation: decoder.take_generation()?,
1944 },
1945 )),
1946 D::MarkerNotDelivered => {
1947 let request = take_marker_proof(required_origin(origin)?, decoder)?;
1948 let reason = take_tag::<t::MarkerNotDeliveredReason>(decoder)?;
1949 Ok(r::ServerValue::MarkerNotDelivered(r::MarkerNotDelivered {
1950 request,
1951 reason,
1952 expected_marker_delivery_seq: decoder.take_u64()?,
1953 }))
1954 }
1955 D::MarkerMismatch => {
1956 let request = take_marker_proof(required_origin(origin)?, decoder)?;
1957 let reason = take_tag::<t::MarkerMismatchReason>(decoder)?;
1958 let mismatch = match reason {
1959 t::MarkerMismatchReason::BelowCursor => r::MarkerMismatchBody::BelowCursor {
1960 current_cursor: decoder.take_u64()?,
1961 },
1962 t::MarkerMismatchReason::NoMarkerExpected => {
1963 r::MarkerMismatchBody::NoMarkerExpected
1964 }
1965 t::MarkerMismatchReason::ExpectedDifferentMarker => {
1966 r::MarkerMismatchBody::ExpectedDifferentMarker {
1967 expected_marker_delivery_seq: decoder.take_u64()?,
1968 }
1969 }
1970 };
1971 Ok(r::ServerValue::MarkerMismatch(r::MarkerMismatch {
1972 request,
1973 mismatch,
1974 }))
1975 }
1976 D::Bound => {
1977 take_receipt_replay(required_origin(origin)?, decoder).map(r::ServerValue::Bound)
1978 }
1979 D::UnboundReceipt => take_receipt_replay(required_origin(origin)?, decoder)
1980 .map(r::ServerValue::UnboundReceipt),
1981 D::DetachCommitted => {
1982 let conversation_id = decoder.take_u64()?;
1983 let participant_id = decoder.take_u64()?;
1984 let capability_generation = decoder.take_generation()?;
1985 let detach_attempt_token = DetachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
1986 let committed_binding_epoch = decoder.take_binding_epoch()?;
1987 let detached_delivery_seq = decoder.take_u64()?;
1988 if capability_generation != committed_binding_epoch.capability_generation {
1989 decoder.invalidate();
1990 }
1991 Ok(r::ServerValue::DetachCommitted(r::DetachCommitted::new(
1992 conversation_id,
1993 participant_id,
1994 detach_attempt_token,
1995 committed_binding_epoch,
1996 detached_delivery_seq,
1997 )))
1998 }
1999 D::DetachInProgress => Ok(r::ServerValue::DetachInProgress(r::DetachInProgress {
2000 conversation_id: decoder.take_u64()?,
2001 participant_id: decoder.take_u64()?,
2002 presented_token: DetachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?),
2003 presented_generation: decoder.take_generation()?,
2004 committed_binding_epoch: decoder.take_binding_epoch()?,
2005 })),
2006 D::AckCommitted => {
2007 let request = take_participant_ack(decoder)?;
2008 if decoder.take_u64()? != request.through_seq {
2009 decoder.invalidate();
2010 }
2011 Ok(r::ServerValue::AckCommitted(r::AckCommitted::new(request)))
2012 }
2013 D::AckNoOp => {
2014 take_ack_no_op(required_origin(origin)?, decoder).map(r::ServerValue::AckNoOp)
2015 }
2016 D::AckGap => {
2017 let request = take_participant_ack(decoder)?;
2018 let current_cursor = decoder.take_u64()?;
2019 let _reason = take_tag::<t::AckGapReason>(decoder)?;
2020 r::AckGap::new(request, current_cursor)
2021 .map(r::ServerValue::AckGap)
2022 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))
2023 }
2024 D::AckRegression => {
2025 let request = take_participant_ack(decoder)?;
2026 let current_cursor = decoder.take_u64()?;
2027 let _reason = take_tag::<t::AckRegressionReason>(decoder)?;
2028 r::AckRegression::new(request, current_cursor)
2029 .map(r::ServerValue::AckRegression)
2030 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))
2031 }
2032 D::LeaveCommitted => {
2033 let conversation_id = decoder.take_u64()?;
2034 let leave_attempt_token = LeaveAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
2035 let participant_id = decoder.take_u64()?;
2036 let presented_generation = decoder.take_generation()?;
2037 let retired_generation = decoder.take_generation()?;
2038 let ended_binding_epoch = decoder.take_option_binding_epoch()?;
2039 let prior_terminal_delivery_seq = decoder.take_option_u64()?;
2040 let left_delivery_seq = decoder.take_u64()?;
2041 if presented_generation != retired_generation {
2042 return Err(decode_error(t::DecodeClass::InvalidField));
2043 }
2044 r::LeaveCommitted::new(
2045 conversation_id,
2046 leave_attempt_token,
2047 participant_id,
2048 retired_generation,
2049 ended_binding_epoch,
2050 prior_terminal_delivery_seq,
2051 left_delivery_seq,
2052 )
2053 .map(r::ServerValue::LeaveCommitted)
2054 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))
2055 }
2056 D::MarkerAckCommitted => {
2057 let request = take_marker_ack(decoder)?;
2058 if decoder.take_u64()? != request.marker_delivery_seq {
2059 decoder.invalidate();
2060 }
2061 Ok(r::ServerValue::MarkerAckCommitted(
2062 r::MarkerAckCommitted::new(request),
2063 ))
2064 }
2065 D::RecordCommitted => {
2066 let request = take_record_admission(decoder)?;
2067 if decoder.take_u64()? != request.participant_id {
2068 decoder.invalidate();
2069 }
2070 let delivery_seq = decoder.take_u64()?;
2071 Ok(r::ServerValue::RecordCommitted(r::RecordCommitted::new(
2072 request,
2073 delivery_seq,
2074 )))
2075 }
2076 D::RecordTooLarge => Ok(r::ServerValue::RecordTooLarge(r::RecordTooLarge {
2077 request: take_record_admission(decoder)?,
2078 dimension: ResourceDimension::from(take_tag::<t::ResourceDimensionTag>(decoder)?),
2079 encoded_record_charge: decoder.take_resource_vector()?,
2080 max_ordinary_record_charge: decoder.take_resource_vector()?,
2081 })),
2082 D::ObserverRecoveryAccepted => {
2083 take_recovery_accepted(decoder).map(r::ServerValue::ObserverRecoveryAccepted)
2084 }
2085 D::InvalidObserverEpoch => {
2086 require_zero_recovery_count(decoder)?;
2087 take_invalid_observer_epoch(decoder).map(r::ServerValue::InvalidObserverEpoch)
2088 }
2089 D::InvalidObserverEpochList => {
2090 require_zero_recovery_count(decoder)?;
2091 take_invalid_observer_epoch_list(decoder).map(r::ServerValue::InvalidObserverEpochList)
2092 }
2093 D::ObserverRecoveryConnectionCapacityExceeded => {
2094 require_zero_recovery_count(decoder)?;
2095 Ok(r::ServerValue::ConnectionConversationCapacityExceeded(
2096 r::ConnectionConversationCapacityExceeded::ObserverRecovery {
2097 conversation_id: decoder.take_u64()?,
2098 limit: decoder.take_u64()?,
2099 },
2100 ))
2101 }
2102 D::MarkerSettlementBackpressure => {
2103 let origin = required_origin(origin)?;
2104 let conversation_id = decoder.take_u64()?;
2105 let refused_epoch = decoder.take_u64()?;
2106 let value = match origin {
2107 t::ClientDiscriminant::CredentialAttachRequest => {
2108 r::MarkerSettlementBackpressure::CredentialAttach {
2109 conversation_id,
2110 refused_epoch,
2111 }
2112 }
2113 t::ClientDiscriminant::DetachRequest => r::MarkerSettlementBackpressure::Detach {
2114 conversation_id,
2115 refused_epoch,
2116 },
2117 _ => return Err(decode_error(t::DecodeClass::InvalidField)),
2118 };
2119 Ok(r::ServerValue::MarkerSettlementBackpressure(value))
2120 }
2121 D::EnrollmentSettlementBackpressure => {
2122 required_origin(origin)?;
2123 Ok(r::ServerValue::EnrollmentSettlementBackpressure(
2124 r::EnrollmentSettlementBackpressure {
2125 conversation_id: decoder.take_u64()?,
2126 },
2127 ))
2128 }
2129 D::RecordAdmissionProtocolFault => Ok(r::ServerValue::RecordAdmissionProtocolFault(
2130 r::RecordAdmissionProtocolFault {
2131 request: take_record_admission(decoder)?,
2132 class: r::RecordAdmissionFaultClass::from(take_tag::<
2133 t::RecordAdmissionFaultClassTag,
2134 >(decoder)?),
2135 },
2136 )),
2137 D::WriteOutcomeUnknown => take_write_outcome_unknown(required_origin(origin)?, decoder)
2138 .map(r::ServerValue::WriteOutcomeUnknown),
2139 }
2140}
2141
2142fn take_write_outcome_unknown(
2143 origin: t::ClientDiscriminant,
2144 decoder: &mut Decoder<'_>,
2145) -> Result<r::WriteOutcomeUnknown, CodecError> {
2146 match origin {
2147 t::ClientDiscriminant::EnrollmentRequest => Ok(r::WriteOutcomeUnknown::Enrollment {
2148 request: take_enrollment(decoder)?,
2149 }),
2150 t::ClientDiscriminant::CredentialAttachRequest => {
2151 Ok(r::WriteOutcomeUnknown::CredentialAttach {
2152 request: take_attach(decoder)?,
2153 })
2154 }
2155 t::ClientDiscriminant::DetachRequest => Ok(r::WriteOutcomeUnknown::Detach {
2156 request: take_detach(decoder)?,
2157 }),
2158 t::ClientDiscriminant::LeaveRequest => Ok(r::WriteOutcomeUnknown::Leave {
2159 request: take_leave(decoder)?,
2160 }),
2161 t::ClientDiscriminant::RecordAdmission => Ok(r::WriteOutcomeUnknown::RecordAdmission {
2162 request: take_record_admission(decoder)?,
2163 }),
2164 _ => Err(decode_error(t::DecodeClass::InvalidField)),
2165 }
2166}
2167
2168trait WireTag: TryFrom<u16> {}
2169
2170impl WireTag for t::TransportReasonTag {}
2171impl WireTag for t::DecodeClass {}
2172impl WireTag for t::AttemptOperation {}
2173impl WireTag for t::AttemptConflict {}
2174impl WireTag for t::Counter {}
2175impl WireTag for t::ClosureScope {}
2176impl WireTag for t::ResourceDimensionTag {}
2177impl WireTag for t::IdentityCapacityScope {}
2178impl WireTag for t::MarkerNotDeliveredReason {}
2179impl WireTag for t::MarkerMismatchReason {}
2180impl WireTag for t::AckGapReason {}
2181impl WireTag for t::AckRegressionReason {}
2182impl WireTag for t::InvalidObserverEpochReason {}
2183impl WireTag for t::InvalidObserverEpochListReason {}
2184impl WireTag for t::RecordAdmissionFaultClassTag {}
2185
2186fn take_tag<T>(decoder: &mut Decoder<'_>) -> Result<T, CodecError>
2187where
2188 T: WireTag,
2189{
2190 T::try_from(decoder.take_u16()?).map_err(|_| decode_error(t::DecodeClass::InvalidField))
2191}
2192
2193fn take_stale_authority(
2194 origin: t::ClientDiscriminant,
2195 decoder: &mut Decoder<'_>,
2196) -> Result<r::StaleAuthority, CodecError> {
2197 match origin {
2198 t::ClientDiscriminant::CredentialAttachRequest => Ok(r::StaleAuthority::Live {
2199 request: r::CommonStaleAuthorityEnvelope::CredentialAttach(take_attach(decoder)?),
2200 current_generation: decoder.take_generation()?,
2201 }),
2202 t::ClientDiscriminant::ParticipantAck => Ok(r::StaleAuthority::Live {
2203 request: r::CommonStaleAuthorityEnvelope::ParticipantAck(take_participant_ack(
2204 decoder,
2205 )?),
2206 current_generation: decoder.take_generation()?,
2207 }),
2208 t::ClientDiscriminant::MarkerAck => Ok(r::StaleAuthority::Live {
2209 request: r::CommonStaleAuthorityEnvelope::MarkerAck(take_marker_ack(decoder)?),
2210 current_generation: decoder.take_generation()?,
2211 }),
2212 t::ClientDiscriminant::RecordAdmission => Ok(r::StaleAuthority::Live {
2213 request: r::CommonStaleAuthorityEnvelope::RecordAdmission(take_record_admission(
2214 decoder,
2215 )?),
2216 current_generation: decoder.take_generation()?,
2217 }),
2218 t::ClientDiscriminant::DetachRequest => {
2219 let authority = t::DetachAuthorityStateTag::try_from(decoder.take_u16()?)
2220 .map_err(|_| decode_error(t::DecodeClass::InvalidField))?;
2221 let value = match authority {
2222 t::DetachAuthorityStateTag::Live => r::DetachStaleAuthority::Live {
2223 conversation_id: decoder.take_u64()?,
2224 participant_id: decoder.take_u64()?,
2225 capability_generation: decoder.take_generation()?,
2226 detach_attempt_token: DetachAttemptToken::new(
2227 decoder.take_fixed::<TOKEN_LEN>()?,
2228 ),
2229 current_generation: decoder.take_generation()?,
2230 },
2231 t::DetachAuthorityStateTag::TerminalizedDetachCell => {
2232 let conversation_id = decoder.take_u64()?;
2233 let participant_id = decoder.take_u64()?;
2234 let capability_generation = decoder.take_generation()?;
2235 let detach_attempt_token =
2236 DetachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
2237 let current_generation = decoder.take_generation()?;
2238 let committed_binding_epoch = decoder.take_binding_epoch()?;
2239 let binding_state = match t::BindingStateTag::try_from(decoder.take_u16()?)
2240 .map_err(|_| decode_error(t::DecodeClass::InvalidField))?
2241 {
2242 t::BindingStateTag::Bound => r::BindingStateView::Bound {
2243 current_binding_epoch: decoder.take_binding_epoch()?,
2244 },
2245 t::BindingStateTag::Detached => r::BindingStateView::Detached,
2246 };
2247 r::DetachStaleAuthority::TerminalizedDetachCell(
2248 r::TerminalizedDetachCell::from_wire_decode(
2249 TERMINALIZED_WIRE_DECODE_AUTHORITY,
2250 conversation_id,
2251 participant_id,
2252 capability_generation,
2253 detach_attempt_token,
2254 current_generation,
2255 committed_binding_epoch,
2256 binding_state,
2257 ),
2258 )
2259 }
2260 };
2261 Ok(r::StaleAuthority::Detach(value))
2262 }
2263 t::ClientDiscriminant::LeaveRequest => {
2264 let authority = t::LeaveAuthorityStateTag::try_from(decoder.take_u16()?)
2265 .map_err(|_| decode_error(t::DecodeClass::InvalidField))?;
2266 let conversation_id = decoder.take_u64()?;
2267 let participant_id = decoder.take_u64()?;
2268 let presented_generation = decoder.take_generation()?;
2269 let leave_attempt_token = LeaveAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
2270 let value = match authority {
2271 t::LeaveAuthorityStateTag::Live => r::LeaveStaleAuthority::Live {
2272 conversation_id,
2273 participant_id,
2274 presented_generation,
2275 leave_attempt_token,
2276 current_generation: decoder.take_generation()?,
2277 },
2278 t::LeaveAuthorityStateTag::CommittedLeaveTombstone => {
2279 r::LeaveStaleAuthority::CommittedLeaveTombstone {
2280 conversation_id,
2281 participant_id,
2282 presented_generation,
2283 leave_attempt_token,
2284 retired_generation: decoder.take_generation()?,
2285 }
2286 }
2287 };
2288 Ok(r::StaleAuthority::Leave(value))
2289 }
2290 t::ClientDiscriminant::EnrollmentRequest
2291 | t::ClientDiscriminant::ObserverRecoveryHandshake => {
2292 Err(decode_error(t::DecodeClass::InvalidField))
2293 }
2294 }
2295}
2296
2297fn take_enroll_bound(decoder: &mut Decoder<'_>) -> Result<r::EnrollBound, CodecError> {
2298 let conversation_id = decoder.take_u64()?;
2299 let token = EnrollmentToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
2300 let participant_id = decoder.take_u64()?;
2301 if decoder.take_option_generation()?.is_some() {
2302 decoder.invalidate();
2303 }
2304 let capability_generation = decoder.take_generation()?;
2305 if capability_generation.get() != 1 {
2306 decoder.invalidate();
2307 }
2308 let attach_secret = AttachSecret::new(decoder.take_fixed::<SECRET_LEN>()?);
2309 let origin_binding_epoch = decoder.take_binding_epoch()?;
2310 if origin_binding_epoch.capability_generation.get() != 1 {
2311 decoder.invalidate();
2312 }
2313 if decoder.take_u64()? != 0 {
2314 decoder.invalidate();
2315 }
2316 if decoder.take_option_u64()?.is_some() {
2317 decoder.invalidate();
2318 }
2319 let receipt_expires_at = decoder.take_u128()?;
2320 let provenance_expires_at = decoder.take_u128()?;
2321
2322 let valid_epoch = if origin_binding_epoch.capability_generation.get() == 1 {
2323 origin_binding_epoch
2324 } else {
2325 BindingEpoch::new(
2326 origin_binding_epoch.connection_incarnation,
2327 generation_one()?,
2328 )
2329 };
2330 r::EnrollBound::new(
2331 conversation_id,
2332 token,
2333 participant_id,
2334 attach_secret,
2335 valid_epoch,
2336 receipt_expires_at,
2337 provenance_expires_at,
2338 )
2339 .ok_or(CodecError::InvalidValue)
2340}
2341
2342fn take_receipt_expired(
2343 origin: t::ClientDiscriminant,
2344 decoder: &mut Decoder<'_>,
2345) -> Result<r::ReceiptExpired, CodecError> {
2346 let conversation_id = decoder.take_u64()?;
2347 let token = decoder.take_fixed::<TOKEN_LEN>()?;
2348 let participant_id = decoder.take_u64()?;
2349 match origin {
2350 t::ClientDiscriminant::EnrollmentRequest => {
2351 if decoder.take_option_generation()?.is_some() {
2352 decoder.invalidate();
2353 }
2354 Ok(r::ReceiptExpired::Enrollment {
2355 conversation_id,
2356 token: EnrollmentToken::new(token),
2357 participant_id,
2358 result_generation: decoder.take_generation()?,
2359 current_generation: decoder.take_generation()?,
2360 reason: take_tag::<t::ReceiptExpiryReason>(decoder)?,
2361 })
2362 }
2363 t::ClientDiscriminant::CredentialAttachRequest => {
2364 let presented_generation = required_generation_option(decoder)?;
2365 Ok(r::ReceiptExpired::CredentialAttach {
2366 conversation_id,
2367 token: AttachAttemptToken::new(token),
2368 participant_id,
2369 presented_generation,
2370 presented_marker_delivery_seq: decoder.take_option_u64()?,
2371 result_generation: decoder.take_generation()?,
2372 current_generation: decoder.take_generation()?,
2373 reason: take_tag::<t::ReceiptExpiryReason>(decoder)?,
2374 })
2375 }
2376 _ => Err(decode_error(t::DecodeClass::InvalidField)),
2377 }
2378}
2379
2380impl WireTag for t::ReceiptExpiryReason {}
2381impl WireTag for t::ReceiptCapacityScope {}
2382
2383fn take_receipt_capacity(
2384 origin: t::ClientDiscriminant,
2385 decoder: &mut Decoder<'_>,
2386) -> Result<r::ReceiptCapacityExceeded, CodecError> {
2387 match origin {
2388 t::ClientDiscriminant::EnrollmentRequest => {
2389 let request = take_enrollment(decoder)?;
2390 let scope = match take_tag::<t::ReceiptCapacityScope>(decoder)? {
2391 t::ReceiptCapacityScope::LiveReceiptServer => {
2392 r::EnrollmentReceiptCapacityScope::LiveReceiptServer
2393 }
2394 t::ReceiptCapacityScope::ProvenanceServer => {
2395 r::EnrollmentReceiptCapacityScope::ProvenanceServer
2396 }
2397 t::ReceiptCapacityScope::ProvenanceConversation => {
2398 r::EnrollmentReceiptCapacityScope::ProvenanceConversation
2399 }
2400 t::ReceiptCapacityScope::LiveReceiptParticipant
2401 | t::ReceiptCapacityScope::ProvenanceParticipant => {
2402 return Err(decode_error(t::DecodeClass::InvalidField));
2403 }
2404 };
2405 let limit = decoder.take_u64()?;
2406 let occupied = decoder.take_u64()?;
2407 require_one(decoder, r::ReceiptCapacityExceeded::REQUESTED)?;
2408 Ok(r::ReceiptCapacityExceeded::Enrollment {
2409 request,
2410 scope,
2411 limit,
2412 occupied,
2413 })
2414 }
2415 t::ClientDiscriminant::CredentialAttachRequest => {
2416 let request = take_attach(decoder)?;
2417 let scope = take_tag::<t::ReceiptCapacityScope>(decoder)?;
2418 let limit = decoder.take_u64()?;
2419 let occupied = decoder.take_u64()?;
2420 require_one(decoder, r::ReceiptCapacityExceeded::REQUESTED)?;
2421 Ok(r::ReceiptCapacityExceeded::CredentialAttach {
2422 request,
2423 scope,
2424 limit,
2425 occupied,
2426 })
2427 }
2428 _ => Err(decode_error(t::DecodeClass::InvalidField)),
2429 }
2430}
2431
2432fn require_one(decoder: &mut Decoder<'_>, expected: u64) -> Result<(), CodecError> {
2433 if decoder.take_u64()? != expected {
2434 decoder.invalidate();
2435 }
2436 Ok(())
2437}
2438
2439fn take_observer_backpressure(
2440 origin: t::ClientDiscriminant,
2441 decoder: &mut Decoder<'_>,
2442) -> Result<r::ObserverBackpressure, CodecError> {
2443 match origin {
2444 t::ClientDiscriminant::EnrollmentRequest => Ok(r::ObserverBackpressure::Enrollment {
2445 request: take_enrollment(decoder)?,
2446 state: take_backpressure_state(decoder)?,
2447 }),
2448 t::ClientDiscriminant::CredentialAttachRequest => {
2449 Ok(r::ObserverBackpressure::CredentialAttach {
2450 request: take_attach(decoder)?,
2451 state: take_backpressure_state(decoder)?,
2452 })
2453 }
2454 t::ClientDiscriminant::DetachRequest => Ok(r::ObserverBackpressure::Detach {
2455 request: take_detach(decoder)?,
2456 committed_binding_epoch: decoder.take_binding_epoch()?,
2457 state: take_backpressure_state(decoder)?,
2458 }),
2459 t::ClientDiscriminant::LeaveRequest => Ok(r::ObserverBackpressure::Leave {
2460 request: take_leave(decoder)?,
2461 state: take_backpressure_state(decoder)?,
2462 prior_terminal_cell_exists: decoder.take_bool()?,
2463 }),
2464 t::ClientDiscriminant::RecordAdmission => Ok(r::ObserverBackpressure::RecordAdmission {
2465 request: take_record_admission(decoder)?,
2466 state: take_backpressure_state(decoder)?,
2467 }),
2468 _ => Err(decode_error(t::DecodeClass::InvalidField)),
2469 }
2470}
2471
2472fn take_backpressure_state(
2473 decoder: &mut Decoder<'_>,
2474) -> Result<r::ObserverBackpressureState, CodecError> {
2475 let backpressure_epoch = decoder.take_u64()?;
2476 let observer_progress = decoder.take_u64()?;
2477 r::ObserverBackpressureState::replay(backpressure_epoch, observer_progress)
2478 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))
2479}
2480
2481fn required_generation_option(decoder: &mut Decoder<'_>) -> Result<Generation, CodecError> {
2482 decoder.take_option_generation()?.map_or_else(
2483 || {
2484 decoder.invalidate();
2485 generation_one()
2486 },
2487 Ok,
2488 )
2489}
2490
2491fn take_attach_bound(decoder: &mut Decoder<'_>) -> Result<r::AttachBound, CodecError> {
2492 let conversation_id = decoder.take_u64()?;
2493 let token = AttachAttemptToken::new(decoder.take_fixed::<TOKEN_LEN>()?);
2494 let participant_id = decoder.take_u64()?;
2495 let request_generation = required_generation_option(decoder)?;
2496 let capability_generation = decoder.take_generation()?;
2497 let attach_secret = AttachSecret::new(decoder.take_fixed::<SECRET_LEN>()?);
2498 let origin_binding_epoch = decoder.take_binding_epoch()?;
2499 let persisted_cursor = decoder.take_u64()?;
2500 let accepted_marker_delivery_seq = decoder.take_option_u64()?;
2501 let receipt_expires_at = decoder.take_u128()?;
2502 let provenance_expires_at = decoder.take_u128()?;
2503
2504 let value = match accepted_marker_delivery_seq {
2505 Some(marker) if persisted_cursor == marker => r::AttachBound::fenced(
2506 conversation_id,
2507 token,
2508 participant_id,
2509 request_generation,
2510 attach_secret,
2511 origin_binding_epoch,
2512 marker,
2513 receipt_expires_at,
2514 provenance_expires_at,
2515 ),
2516 Some(_) => None,
2517 None => r::AttachBound::ordinary(
2518 conversation_id,
2519 token,
2520 participant_id,
2521 request_generation,
2522 attach_secret,
2523 origin_binding_epoch,
2524 persisted_cursor,
2525 receipt_expires_at,
2526 provenance_expires_at,
2527 ),
2528 }
2529 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))?;
2530 if value.capability_generation() == capability_generation {
2531 Ok(value)
2532 } else {
2533 Err(decode_error(t::DecodeClass::InvalidField))
2534 }
2535}
2536
2537fn take_receipt_replay(
2538 origin: t::ClientDiscriminant,
2539 decoder: &mut Decoder<'_>,
2540) -> Result<r::ReceiptReplay, CodecError> {
2541 let conversation_id = decoder.take_u64()?;
2542 let token = decoder.take_fixed::<TOKEN_LEN>()?;
2543 let participant_id = decoder.take_u64()?;
2544 match origin {
2545 t::ClientDiscriminant::EnrollmentRequest => {
2546 let request_generation = decoder.take_option_generation()?;
2547 let capability_generation = decoder.take_generation()?;
2548 let attach_secret = AttachSecret::new(decoder.take_fixed::<SECRET_LEN>()?);
2549 let origin_binding_epoch = decoder.take_binding_epoch()?;
2550 let persisted_cursor = decoder.take_u64()?;
2551 let accepted_marker_delivery_seq = decoder.take_option_u64()?;
2552 let receipt_expires_at = decoder.take_u128()?;
2553 let provenance_expires_at = decoder.take_u128()?;
2554 let value = r::EnrollBound::new(
2555 conversation_id,
2556 EnrollmentToken::new(token),
2557 participant_id,
2558 attach_secret,
2559 origin_binding_epoch,
2560 receipt_expires_at,
2561 provenance_expires_at,
2562 )
2563 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))?;
2564 if request_generation.is_none()
2565 && capability_generation == Generation::ONE
2566 && persisted_cursor == 0
2567 && accepted_marker_delivery_seq.is_none()
2568 {
2569 Ok(r::ReceiptReplay::Enrollment(value))
2570 } else {
2571 Err(decode_error(t::DecodeClass::InvalidField))
2572 }
2573 }
2574 t::ClientDiscriminant::CredentialAttachRequest => {
2575 let request_generation = required_generation_option(decoder)?;
2576 let capability_generation = decoder.take_generation()?;
2577 let attach_secret = AttachSecret::new(decoder.take_fixed::<SECRET_LEN>()?);
2578 let origin_binding_epoch = decoder.take_binding_epoch()?;
2579 let persisted_cursor = decoder.take_u64()?;
2580 let accepted_marker_delivery_seq = decoder.take_option_u64()?;
2581 let receipt_expires_at = decoder.take_u128()?;
2582 let provenance_expires_at = decoder.take_u128()?;
2583 let attach_token = AttachAttemptToken::new(token);
2584 let value = match accepted_marker_delivery_seq {
2585 Some(marker) if persisted_cursor == marker => r::AttachBound::fenced(
2586 conversation_id,
2587 attach_token,
2588 participant_id,
2589 request_generation,
2590 attach_secret,
2591 origin_binding_epoch,
2592 marker,
2593 receipt_expires_at,
2594 provenance_expires_at,
2595 ),
2596 Some(_) => None,
2597 None => r::AttachBound::ordinary(
2598 conversation_id,
2599 attach_token,
2600 participant_id,
2601 request_generation,
2602 attach_secret,
2603 origin_binding_epoch,
2604 persisted_cursor,
2605 receipt_expires_at,
2606 provenance_expires_at,
2607 ),
2608 }
2609 .ok_or_else(|| decode_error(t::DecodeClass::InvalidField))?;
2610 if value.capability_generation() == capability_generation {
2611 Ok(r::ReceiptReplay::CredentialAttach(value))
2612 } else {
2613 Err(decode_error(t::DecodeClass::InvalidField))
2614 }
2615 }
2616 _ => Err(decode_error(t::DecodeClass::InvalidField)),
2617 }
2618}
2619
2620fn take_ack_no_op(
2621 origin: t::ClientDiscriminant,
2622 decoder: &mut Decoder<'_>,
2623) -> Result<r::AckNoOp, CodecError> {
2624 match origin {
2625 t::ClientDiscriminant::ParticipantAck => {
2626 let request = take_participant_ack(decoder)?;
2627 if decoder.take_u64()? != request.through_seq {
2628 decoder.invalidate();
2629 }
2630 Ok(r::AckNoOp::participant_ack(request))
2631 }
2632 t::ClientDiscriminant::MarkerAck => {
2633 let request = take_marker_ack(decoder)?;
2634 if decoder.take_u64()? != request.marker_delivery_seq {
2635 decoder.invalidate();
2636 }
2637 Ok(r::AckNoOp::marker_ack(request))
2638 }
2639 _ => Err(decode_error(t::DecodeClass::InvalidField)),
2640 }
2641}
2642
2643fn take_recovery_accepted(
2644 decoder: &mut Decoder<'_>,
2645) -> Result<r::ObserverRecoveryAccepted, CodecError> {
2646 let count = decoder.take_u64()?;
2647 let required = u128::from(count) * RECOVERY_STATUS_LEN;
2648 let remaining = u64::try_from(decoder.remaining())
2649 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
2650 if required > u128::from(remaining) {
2651 return Err(decode_error(t::DecodeClass::MissingRequiredField));
2652 }
2653 let count_usize: usize = count
2654 .try_into()
2655 .map_err(|_| decode_error(t::DecodeClass::MissingRequiredField))?;
2656
2657 let mut statuses = Vec::new();
2658 for _ in 0..count_usize {
2659 statuses.push(r::ObserverProgressStatus {
2660 conversation_id: decoder.take_u64()?,
2661 refused_epoch: decoder.take_u64()?,
2662 current_observer_progress: decoder.take_u64()?,
2663 armed: decoder.take_bool()?,
2664 progressed: decoder.take_bool()?,
2665 });
2666 }
2667 Ok(r::ObserverRecoveryAccepted { statuses })
2668}
2669
2670fn require_zero_recovery_count(decoder: &mut Decoder<'_>) -> Result<(), CodecError> {
2671 if decoder.take_u64()? == 0 {
2672 Ok(())
2673 } else {
2674 Err(decode_error(t::DecodeClass::InvalidField))
2675 }
2676}
2677
2678fn take_invalid_observer_epoch(
2679 decoder: &mut Decoder<'_>,
2680) -> Result<r::InvalidObserverEpoch, CodecError> {
2681 let reason = take_tag::<t::InvalidObserverEpochReason>(decoder)?;
2682 let conversation_id = decoder.take_u64()?;
2683 let presented_epoch = decoder.take_u64()?;
2684 let current = decoder.take_option_u64()?;
2685 match reason {
2686 t::InvalidObserverEpochReason::ConversationUnknown => {
2687 if current.is_some() {
2688 decoder.invalidate();
2689 }
2690 Ok(r::InvalidObserverEpoch::ConversationUnknown {
2691 conversation_id,
2692 presented_epoch,
2693 })
2694 }
2695 t::InvalidObserverEpochReason::EpochAhead => {
2696 let current_observer_progress = current.unwrap_or_else(|| {
2697 decoder.invalidate();
2698 0
2699 });
2700 Ok(r::InvalidObserverEpoch::EpochAhead {
2701 conversation_id,
2702 presented_epoch,
2703 current_observer_progress,
2704 })
2705 }
2706 }
2707}
2708
2709fn take_invalid_observer_epoch_list(
2710 decoder: &mut Decoder<'_>,
2711) -> Result<r::InvalidObserverEpochList, CodecError> {
2712 match take_tag::<t::InvalidObserverEpochListReason>(decoder)? {
2713 t::InvalidObserverEpochListReason::TooManyEntries => {
2714 Ok(r::InvalidObserverEpochList::TooManyEntries {
2715 presented_entries: decoder.take_u64()?,
2716 max_entries: decoder.take_u64()?,
2717 })
2718 }
2719 t::InvalidObserverEpochListReason::DuplicateConversation => {
2720 Ok(r::InvalidObserverEpochList::DuplicateConversation {
2721 conversation_id: decoder.take_u64()?,
2722 first_index: decoder.take_u64()?,
2723 duplicate_index: decoder.take_u64()?,
2724 })
2725 }
2726 }
2727}