microsandbox_protocol/
bulk.rs

1//! Generation-8 raw bulk records, control payloads, and flow state.
2
3use bytes::Bytes;
4use serde::{Deserialize, Deserializer, Serialize, Serializer};
5use thiserror::Error;
6
7//--------------------------------------------------------------------------------------------------
8// Constants
9//--------------------------------------------------------------------------------------------------
10
11/// Generation that introduced raw bulk records.
12pub const BULK_PROTOCOL_VERSION: u8 = 8;
13
14/// First and only bulk record format in generation 8.
15pub const BULK_FORMAT_RAW_V1: u8 = 1;
16
17/// Bytes in the raw record body before its payload.
18pub const BULK_HEADER_SIZE: usize = 12;
19
20/// Default raw-record payload selected by generation-8 peers.
21pub const DEFAULT_BULK_RECORD_PAYLOAD: u32 = 256 * 1024;
22
23/// Default filesystem payload selected by generation-8 peers.
24///
25/// This matches the existing filesystem streaming chunk so raw transport removes serialization
26/// work without multiplying file, channel and scheduler operations. TCP retains the smaller
27/// default above for interactive latency and fairness.
28pub const DEFAULT_FILESYSTEM_BULK_RECORD_PAYLOAD: u32 = 3 * 1024 * 1024;
29
30/// Smallest record payload a peer may negotiate.
31pub const MIN_BULK_RECORD_PAYLOAD: u32 = 16 * 1024;
32
33/// Largest record payload generation 8 permits.
34pub const MAX_BULK_RECORD_PAYLOAD: u32 = DEFAULT_FILESYSTEM_BULK_RECORD_PAYLOAD;
35
36/// Default receive window granted to one bulk flow.
37pub const DEFAULT_BULK_WINDOW: u64 = 8 * 1024 * 1024;
38
39/// Largest receive window generation 8 permits.
40pub const MAX_BULK_WINDOW: u64 = 32 * 1024 * 1024;
41
42/// Flow-mask bit for host-to-guest data.
43pub const BULK_FLOW_MASK_HOST_TO_GUEST: u8 = 0b01;
44
45/// Flow-mask bit for guest-to-host data.
46pub const BULK_FLOW_MASK_GUEST_TO_HOST: u8 = 0b10;
47
48//--------------------------------------------------------------------------------------------------
49// Types
50//--------------------------------------------------------------------------------------------------
51
52/// Operation family carried by a raw bulk record.
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54#[repr(u8)]
55pub enum BulkKind {
56    /// Filesystem read or write bytes.
57    Filesystem = 1,
58
59    /// TCP stream bytes.
60    Tcp = 2,
61}
62
63/// Physical direction of a raw bulk flow across the VM boundary.
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65#[repr(u8)]
66pub enum BulkFlow {
67    /// Host or SDK to guest or agentd.
68    HostToGuest = 1,
69
70    /// Guest or agentd to host or SDK.
71    GuestToHost = 2,
72}
73
74/// A generation-8 raw record after fixed-header validation.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub struct BulkRecord {
77    /// Correlation that owns this record.
78    pub id: u32,
79
80    /// Operation family.
81    pub kind: BulkKind,
82
83    /// Physical data direction.
84    pub flow: BulkFlow,
85
86    /// Zero-based stream offset of the first payload byte.
87    pub offset: u64,
88
89    /// Opaque payload bytes.
90    pub payload: Bytes,
91}
92
93/// Optional raw-bulk offer on a generation-8 opening request.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
95pub struct BulkOffer {
96    /// Bulk record format. Generation 8 requires `1`.
97    pub format: u8,
98
99    /// Largest payload this host accepts in one raw record.
100    pub max_record_payload: u32,
101
102    /// Initial absolute guest-to-host exclusive send limit.
103    pub guest_to_host_credit_limit: u64,
104}
105
106/// Guest acceptance of one offered bulk correlation.
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
108pub struct BulkAccepted {
109    /// Accepted operation family.
110    pub kind: BulkKind,
111
112    /// Enabled-flow mask.
113    pub flows: u8,
114
115    /// Accepted record format.
116    pub format: u8,
117
118    /// Effective maximum record payload.
119    pub max_record_payload: u32,
120
121    /// Initial absolute host-to-guest exclusive send limit.
122    pub host_to_guest_credit_limit: u64,
123
124    /// Exact host grant accepted for guest-to-host data.
125    pub guest_to_host_credit_limit: u64,
126}
127
128/// Absolute credit update from a flow receiver.
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
130pub struct BulkCredit {
131    /// Operation family.
132    pub kind: BulkKind,
133
134    /// Flow whose sender receives credit.
135    pub flow: BulkFlow,
136
137    /// Bytes for which the receiver has accepted responsibility.
138    pub consumed_offset: u64,
139
140    /// Absolute exclusive byte offset the sender may reach.
141    pub credit_limit: u64,
142}
143
144/// Exact end offset for one flow.
145#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
146pub struct BulkFinish {
147    /// Operation family.
148    pub kind: BulkKind,
149
150    /// Flow that reached EOF or half-close.
151    pub flow: BulkFlow,
152
153    /// Exact final byte offset.
154    pub final_offset: u64,
155}
156
157/// Reason one peer cancelled an entire bulk correlation.
158#[derive(Debug, Clone, Copy, PartialEq, Eq)]
159#[repr(u8)]
160pub enum BulkCancelReason {
161    /// Local caller dropped or cancelled the operation.
162    CallerCancelled = 1,
163
164    /// Destination file or socket I/O failed.
165    DestinationIo = 2,
166
167    /// A configured resource limit was reached.
168    ResourceLimit = 3,
169
170    /// The underlying transport failed.
171    TransportFailure = 4,
172
173    /// The peer violated the correlation state machine.
174    ProtocolState = 5,
175}
176
177/// Best-effort request to stop an entire bulk correlation.
178#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
179pub struct BulkCancel {
180    /// Operation family.
181    pub kind: BulkKind,
182
183    /// Stable cancellation reason.
184    pub reason: BulkCancelReason,
185
186    /// Human-readable diagnostic without payload data.
187    pub message: String,
188}
189
190/// Sender-side absolute-offset and credit state for one flow.
191#[derive(Debug, Clone)]
192pub struct BulkSendState {
193    kind: BulkKind,
194    flow: BulkFlow,
195    max_record_payload: u32,
196    next_offset: u64,
197    consumed_offset: u64,
198    credit_limit: u64,
199    finished: bool,
200}
201
202/// Receiver-side exact-offset and replenishment state for one flow.
203#[derive(Debug, Clone)]
204pub struct BulkReceiveState {
205    kind: BulkKind,
206    flow: BulkFlow,
207    max_record_payload: u32,
208    window: u64,
209    next_expected_offset: u64,
210    consumed_offset: u64,
211    credit_limit: u64,
212    finished: bool,
213}
214
215/// Validation failure in generation-8 bulk state.
216#[derive(Debug, Clone, PartialEq, Eq, Error)]
217pub enum BulkStateError {
218    /// Unsupported record format.
219    #[error("unsupported bulk format {0}")]
220    UnsupportedFormat(u8),
221
222    /// Negotiated record limit is outside generation-8 bounds.
223    #[error("invalid maximum bulk record payload {0}")]
224    InvalidRecordLimit(u32),
225
226    /// A raw record payload is empty or exceeds the negotiated limit.
227    #[error("invalid bulk record payload length {length} (max {max})")]
228    InvalidPayloadLength {
229        /// Actual payload length.
230        length: usize,
231        /// Negotiated maximum.
232        max: u32,
233    },
234
235    /// The record or control belongs to another operation family or flow.
236    #[error("bulk kind or flow does not match the correlation")]
237    FlowMismatch,
238
239    /// Offset arithmetic overflowed `u64`.
240    #[error("bulk offset overflow")]
241    OffsetOverflow,
242
243    /// A sender tried to exceed the absolute receive credit.
244    #[error("bulk record end {end} exceeds credit limit {limit}")]
245    CreditExceeded {
246        /// Proposed exclusive record end.
247        end: u64,
248        /// Current exclusive credit limit.
249        limit: u64,
250    },
251
252    /// Credit state regressed or violated the negotiated window.
253    #[error("invalid bulk credit: {0}")]
254    InvalidCredit(String),
255
256    /// A record did not start at the exact expected stream offset.
257    #[error("bulk record offset {actual} does not match expected {expected}")]
258    OffsetMismatch {
259        /// Required offset.
260        expected: u64,
261        /// Received offset.
262        actual: u64,
263    },
264
265    /// A finish marker did not match the exact admitted offset.
266    #[error("bulk finish offset {actual} does not match expected {expected}")]
267    FinishMismatch {
268        /// Required final offset.
269        expected: u64,
270        /// Received final offset.
271        actual: u64,
272    },
273
274    /// Data or control arrived after this flow finished.
275    #[error("bulk flow is already finished")]
276    AlreadyFinished,
277}
278
279//--------------------------------------------------------------------------------------------------
280// Methods
281//--------------------------------------------------------------------------------------------------
282
283impl BulkKind {
284    /// Parse a generation-8 wire value.
285    pub fn from_wire(value: u8) -> Option<Self> {
286        match value {
287            1 => Some(Self::Filesystem),
288            2 => Some(Self::Tcp),
289            _ => None,
290        }
291    }
292}
293
294impl BulkFlow {
295    /// Parse a generation-8 wire value.
296    pub fn from_wire(value: u8) -> Option<Self> {
297        match value {
298            1 => Some(Self::HostToGuest),
299            2 => Some(Self::GuestToHost),
300            _ => None,
301        }
302    }
303
304    /// Return this flow's bit in [`BulkAccepted::flows`].
305    pub fn mask(self) -> u8 {
306        match self {
307            Self::HostToGuest => BULK_FLOW_MASK_HOST_TO_GUEST,
308            Self::GuestToHost => BULK_FLOW_MASK_GUEST_TO_HOST,
309        }
310    }
311}
312
313impl BulkOffer {
314    /// Build the default offer for a filesystem read.
315    pub fn filesystem_read() -> Self {
316        Self::new(DEFAULT_FILESYSTEM_BULK_RECORD_PAYLOAD, DEFAULT_BULK_WINDOW)
317    }
318
319    /// Build the default offer for a filesystem write.
320    pub fn filesystem_write() -> Self {
321        Self::new(DEFAULT_FILESYSTEM_BULK_RECORD_PAYLOAD, 0)
322    }
323
324    /// Build the default bidirectional TCP offer.
325    pub fn tcp() -> Self {
326        Self::new(DEFAULT_BULK_RECORD_PAYLOAD, DEFAULT_BULK_WINDOW)
327    }
328
329    /// Validate generation-8 offer limits.
330    pub fn validate(self) -> Result<Self, BulkStateError> {
331        validate_record_limit(self.max_record_payload)?;
332        if self.format != BULK_FORMAT_RAW_V1 {
333            return Err(BulkStateError::UnsupportedFormat(self.format));
334        }
335        if self.guest_to_host_credit_limit > MAX_BULK_WINDOW {
336            return Err(BulkStateError::InvalidCredit(format!(
337                "initial guest-to-host limit {} exceeds {}",
338                self.guest_to_host_credit_limit, MAX_BULK_WINDOW
339            )));
340        }
341        Ok(self)
342    }
343
344    fn new(max_record_payload: u32, guest_to_host_credit_limit: u64) -> Self {
345        Self {
346            format: BULK_FORMAT_RAW_V1,
347            max_record_payload,
348            guest_to_host_credit_limit,
349        }
350    }
351}
352
353impl BulkAccepted {
354    /// Validate an acceptance against the opening offer and required operation shape.
355    pub fn validate_against(
356        self,
357        offer: BulkOffer,
358        kind: BulkKind,
359        flows: u8,
360    ) -> Result<Self, BulkStateError> {
361        let offer = offer.validate()?;
362        validate_record_limit(self.max_record_payload)?;
363        if self.kind != kind || self.flows != flows || self.flows & !0b11 != 0 {
364            return Err(BulkStateError::FlowMismatch);
365        }
366        if self.format != BULK_FORMAT_RAW_V1 {
367            return Err(BulkStateError::UnsupportedFormat(self.format));
368        }
369        if self.max_record_payload > offer.max_record_payload {
370            return Err(BulkStateError::InvalidRecordLimit(self.max_record_payload));
371        }
372        if self.guest_to_host_credit_limit != offer.guest_to_host_credit_limit {
373            return Err(BulkStateError::InvalidCredit(
374                "guest-to-host grant was not echoed exactly".into(),
375            ));
376        }
377        if self.host_to_guest_credit_limit > MAX_BULK_WINDOW {
378            return Err(BulkStateError::InvalidCredit(
379                "host-to-guest grant exceeds the generation-8 window".into(),
380            ));
381        }
382        let host_to_guest = self.flows & BULK_FLOW_MASK_HOST_TO_GUEST != 0;
383        let guest_to_host = self.flows & BULK_FLOW_MASK_GUEST_TO_HOST != 0;
384        if host_to_guest != (self.host_to_guest_credit_limit != 0) {
385            return Err(BulkStateError::InvalidCredit(
386                "host-to-guest credit must be nonzero exactly when that flow is enabled".into(),
387            ));
388        }
389        if guest_to_host != (self.guest_to_host_credit_limit != 0) {
390            return Err(BulkStateError::InvalidCredit(
391                "guest-to-host credit must be nonzero exactly when that flow is enabled".into(),
392            ));
393        }
394        Ok(self)
395    }
396}
397
398impl BulkSendState {
399    /// Create one sender flow from negotiated limits.
400    pub fn new(
401        kind: BulkKind,
402        flow: BulkFlow,
403        max_record_payload: u32,
404        credit_limit: u64,
405    ) -> Result<Self, BulkStateError> {
406        validate_record_limit(max_record_payload)?;
407        if credit_limit > MAX_BULK_WINDOW {
408            return Err(BulkStateError::InvalidCredit(
409                "initial credit exceeds the generation-8 window".into(),
410            ));
411        }
412        Ok(Self {
413            kind,
414            flow,
415            max_record_payload,
416            next_offset: 0,
417            consumed_offset: 0,
418            credit_limit,
419            finished: false,
420        })
421    }
422
423    /// Return the next byte offset this sender will assign.
424    pub fn next_offset(&self) -> u64 {
425        self.next_offset
426    }
427
428    /// Return the currently admitted exclusive limit.
429    pub fn credit_limit(&self) -> u64 {
430        self.credit_limit
431    }
432
433    /// Return the maximum payload negotiated for this flow.
434    pub fn max_record_payload(&self) -> u32 {
435        self.max_record_payload
436    }
437
438    /// Return bytes the sender may currently admit without another credit update.
439    pub fn available_credit(&self) -> u64 {
440        self.credit_limit.saturating_sub(self.next_offset)
441    }
442
443    /// Admit a payload and advance the ordered sender cursor.
444    pub fn admit(&mut self, payload_len: usize) -> Result<u64, BulkStateError> {
445        if self.finished {
446            return Err(BulkStateError::AlreadyFinished);
447        }
448        validate_payload_len(payload_len, self.max_record_payload)?;
449        let end = self
450            .next_offset
451            .checked_add(payload_len as u64)
452            .ok_or(BulkStateError::OffsetOverflow)?;
453        if end > self.credit_limit {
454            return Err(BulkStateError::CreditExceeded {
455                end,
456                limit: self.credit_limit,
457            });
458        }
459        let offset = self.next_offset;
460        self.next_offset = end;
461        Ok(offset)
462    }
463
464    /// Apply one idempotent absolute credit update.
465    pub fn apply_credit(&mut self, credit: BulkCredit) -> Result<bool, BulkStateError> {
466        if credit.kind != self.kind || credit.flow != self.flow {
467            return Err(BulkStateError::FlowMismatch);
468        }
469        if credit.consumed_offset > self.next_offset {
470            return Err(BulkStateError::InvalidCredit(
471                "peer consumed bytes the sender has not admitted".into(),
472            ));
473        }
474        if credit.credit_limit < credit.consumed_offset
475            || credit.credit_limit - credit.consumed_offset > MAX_BULK_WINDOW
476        {
477            return Err(BulkStateError::InvalidCredit(
478                "credit limit is outside the allowed absolute window".into(),
479            ));
480        }
481
482        if credit.consumed_offset <= self.consumed_offset
483            && credit.credit_limit <= self.credit_limit
484        {
485            return Ok(false);
486        }
487        if credit.consumed_offset < self.consumed_offset
488            || credit.credit_limit < self.credit_limit
489            || credit.credit_limit < self.next_offset
490        {
491            return Err(BulkStateError::InvalidCredit(
492                "credit fields advanced inconsistently".into(),
493            ));
494        }
495
496        self.consumed_offset = credit.consumed_offset;
497        self.credit_limit = credit.credit_limit;
498        Ok(true)
499    }
500
501    /// Finish this sender at its exact current offset.
502    pub fn finish(&mut self) -> Result<BulkFinish, BulkStateError> {
503        if self.finished {
504            return Err(BulkStateError::AlreadyFinished);
505        }
506        self.finished = true;
507        Ok(BulkFinish {
508            kind: self.kind,
509            flow: self.flow,
510            final_offset: self.next_offset,
511        })
512    }
513}
514
515impl BulkReceiveState {
516    /// Create one receiver flow and its initial absolute grant.
517    pub fn new(
518        kind: BulkKind,
519        flow: BulkFlow,
520        max_record_payload: u32,
521        credit_limit: u64,
522        window: u64,
523    ) -> Result<Self, BulkStateError> {
524        validate_record_limit(max_record_payload)?;
525        if window == 0 || window > MAX_BULK_WINDOW || credit_limit > window {
526            return Err(BulkStateError::InvalidCredit(
527                "invalid initial receive window".into(),
528            ));
529        }
530        Ok(Self {
531            kind,
532            flow,
533            max_record_payload,
534            window,
535            next_expected_offset: 0,
536            consumed_offset: 0,
537            credit_limit,
538            finished: false,
539        })
540    }
541
542    /// Return the next exact record offset.
543    pub fn next_expected_offset(&self) -> u64 {
544        self.next_expected_offset
545    }
546
547    /// Return the current absolute receive credit limit.
548    pub fn credit_limit(&self) -> u64 {
549        self.credit_limit
550    }
551
552    /// Validate and admit a record before its destination consumes the payload.
553    pub fn accept_record(&mut self, record: &BulkRecord) -> Result<u64, BulkStateError> {
554        if self.finished {
555            return Err(BulkStateError::AlreadyFinished);
556        }
557        if record.kind != self.kind || record.flow != self.flow {
558            return Err(BulkStateError::FlowMismatch);
559        }
560        validate_payload_len(record.payload.len(), self.max_record_payload)?;
561        if record.offset != self.next_expected_offset {
562            return Err(BulkStateError::OffsetMismatch {
563                expected: self.next_expected_offset,
564                actual: record.offset,
565            });
566        }
567        let end = record
568            .offset
569            .checked_add(record.payload.len() as u64)
570            .ok_or(BulkStateError::OffsetOverflow)?;
571        if end > self.credit_limit {
572            return Err(BulkStateError::CreditExceeded {
573                end,
574                limit: self.credit_limit,
575            });
576        }
577        self.next_expected_offset = end;
578        Ok(end)
579    }
580
581    /// Mark admitted bytes consumed and return a replenishment when half the window remains.
582    pub fn consume(&mut self, consumed_offset: u64) -> Result<Option<BulkCredit>, BulkStateError> {
583        if consumed_offset < self.consumed_offset || consumed_offset > self.next_expected_offset {
584            return Err(BulkStateError::InvalidCredit(
585                "consumed offset is outside admitted bytes".into(),
586            ));
587        }
588        self.consumed_offset = consumed_offset;
589        if self.credit_limit - self.consumed_offset > self.window / 2 {
590            return Ok(None);
591        }
592        let next_limit = self
593            .consumed_offset
594            .checked_add(self.window)
595            .ok_or(BulkStateError::OffsetOverflow)?;
596        if next_limit <= self.credit_limit {
597            return Ok(None);
598        }
599        self.credit_limit = next_limit;
600        Ok(Some(BulkCredit {
601            kind: self.kind,
602            flow: self.flow,
603            consumed_offset: self.consumed_offset,
604            credit_limit: self.credit_limit,
605        }))
606    }
607
608    /// Accept an exact finish after all earlier records.
609    pub fn accept_finish(&mut self, finish: BulkFinish) -> Result<(), BulkStateError> {
610        if self.finished {
611            return Err(BulkStateError::AlreadyFinished);
612        }
613        if finish.kind != self.kind || finish.flow != self.flow {
614            return Err(BulkStateError::FlowMismatch);
615        }
616        if finish.final_offset != self.next_expected_offset {
617            return Err(BulkStateError::FinishMismatch {
618                expected: self.next_expected_offset,
619                actual: finish.final_offset,
620            });
621        }
622        self.finished = true;
623        Ok(())
624    }
625}
626
627//--------------------------------------------------------------------------------------------------
628// Trait Implementations
629//--------------------------------------------------------------------------------------------------
630
631macro_rules! impl_wire_enum {
632    ($type:ty, $parse:path) => {
633        impl Serialize for $type {
634            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
635            where
636                S: Serializer,
637            {
638                serializer.serialize_u8(*self as u8)
639            }
640        }
641
642        impl<'de> Deserialize<'de> for $type {
643            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
644            where
645                D: Deserializer<'de>,
646            {
647                let value = u8::deserialize(deserializer)?;
648                $parse(value).ok_or_else(|| serde::de::Error::custom("unknown bulk enum value"))
649            }
650        }
651    };
652}
653
654impl_wire_enum!(BulkKind, BulkKind::from_wire);
655impl_wire_enum!(BulkFlow, BulkFlow::from_wire);
656
657impl Serialize for BulkCancelReason {
658    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
659    where
660        S: Serializer,
661    {
662        serializer.serialize_u8(*self as u8)
663    }
664}
665
666impl<'de> Deserialize<'de> for BulkCancelReason {
667    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
668    where
669        D: Deserializer<'de>,
670    {
671        let value = u8::deserialize(deserializer)?;
672        match value {
673            1 => Ok(Self::CallerCancelled),
674            2 => Ok(Self::DestinationIo),
675            3 => Ok(Self::ResourceLimit),
676            4 => Ok(Self::TransportFailure),
677            5 => Ok(Self::ProtocolState),
678            _ => Err(serde::de::Error::custom("unknown bulk cancel reason")),
679        }
680    }
681}
682
683//--------------------------------------------------------------------------------------------------
684// Functions
685//--------------------------------------------------------------------------------------------------
686
687fn validate_record_limit(limit: u32) -> Result<(), BulkStateError> {
688    if !(MIN_BULK_RECORD_PAYLOAD..=MAX_BULK_RECORD_PAYLOAD).contains(&limit) {
689        return Err(BulkStateError::InvalidRecordLimit(limit));
690    }
691    Ok(())
692}
693
694fn validate_payload_len(length: usize, max: u32) -> Result<(), BulkStateError> {
695    if length == 0 || length > max as usize {
696        return Err(BulkStateError::InvalidPayloadLength { length, max });
697    }
698    Ok(())
699}
700
701//--------------------------------------------------------------------------------------------------
702// Tests
703//--------------------------------------------------------------------------------------------------
704
705#[cfg(test)]
706mod tests {
707    use super::*;
708
709    #[test]
710    fn default_offers_keep_filesystem_throughput_and_tcp_latency_granularity() {
711        assert_eq!(
712            BulkOffer::filesystem_read().max_record_payload,
713            DEFAULT_FILESYSTEM_BULK_RECORD_PAYLOAD
714        );
715        assert_eq!(
716            BulkOffer::filesystem_write().max_record_payload,
717            DEFAULT_FILESYSTEM_BULK_RECORD_PAYLOAD
718        );
719        assert_eq!(
720            BulkOffer::tcp().max_record_payload,
721            DEFAULT_BULK_RECORD_PAYLOAD
722        );
723    }
724
725    #[test]
726    fn new_filesystem_offer_accepts_an_older_peers_smaller_limit() {
727        let offer = BulkOffer::filesystem_write();
728        let accepted = BulkAccepted {
729            kind: BulkKind::Filesystem,
730            flows: BULK_FLOW_MASK_HOST_TO_GUEST,
731            format: BULK_FORMAT_RAW_V1,
732            max_record_payload: DEFAULT_BULK_RECORD_PAYLOAD,
733            host_to_guest_credit_limit: DEFAULT_BULK_WINDOW,
734            guest_to_host_credit_limit: 0,
735        };
736
737        let negotiated = accepted
738            .validate_against(offer, BulkKind::Filesystem, BULK_FLOW_MASK_HOST_TO_GUEST)
739            .unwrap();
740        assert_eq!(negotiated.max_record_payload, DEFAULT_BULK_RECORD_PAYLOAD);
741    }
742
743    #[test]
744    fn sender_stops_at_exact_credit_and_accepts_absolute_replenishment() {
745        let mut sender = BulkSendState::new(
746            BulkKind::Filesystem,
747            BulkFlow::GuestToHost,
748            MIN_BULK_RECORD_PAYLOAD,
749            MIN_BULK_RECORD_PAYLOAD as u64,
750        )
751        .unwrap();
752        assert_eq!(sender.admit(MIN_BULK_RECORD_PAYLOAD as usize).unwrap(), 0);
753        assert!(matches!(
754            sender.admit(1),
755            Err(BulkStateError::CreditExceeded { .. })
756        ));
757
758        assert!(
759            sender
760                .apply_credit(BulkCredit {
761                    kind: BulkKind::Filesystem,
762                    flow: BulkFlow::GuestToHost,
763                    consumed_offset: MIN_BULK_RECORD_PAYLOAD as u64,
764                    credit_limit: 2 * MIN_BULK_RECORD_PAYLOAD as u64,
765                })
766                .unwrap()
767        );
768        assert_eq!(
769            sender.admit(MIN_BULK_RECORD_PAYLOAD as usize).unwrap(),
770            MIN_BULK_RECORD_PAYLOAD as u64
771        );
772    }
773
774    #[test]
775    fn receiver_rejects_gap_and_replenishes_at_half_window() {
776        let window = 2 * MIN_BULK_RECORD_PAYLOAD as u64;
777        let mut receiver = BulkReceiveState::new(
778            BulkKind::Tcp,
779            BulkFlow::HostToGuest,
780            MIN_BULK_RECORD_PAYLOAD,
781            window,
782            window,
783        )
784        .unwrap();
785        let gap = BulkRecord {
786            id: 4,
787            kind: BulkKind::Tcp,
788            flow: BulkFlow::HostToGuest,
789            offset: 1,
790            payload: Bytes::from_static(b"x"),
791        };
792        assert!(matches!(
793            receiver.accept_record(&gap),
794            Err(BulkStateError::OffsetMismatch { .. })
795        ));
796
797        let record = BulkRecord {
798            offset: 0,
799            payload: Bytes::from(vec![0; MIN_BULK_RECORD_PAYLOAD as usize]),
800            ..gap
801        };
802        let end = receiver.accept_record(&record).unwrap();
803        let credit = receiver.consume(end).unwrap().unwrap();
804        assert_eq!(credit.consumed_offset, MIN_BULK_RECORD_PAYLOAD as u64);
805        assert_eq!(credit.credit_limit, 3 * MIN_BULK_RECORD_PAYLOAD as u64);
806    }
807}
microsandbox_protocol/bulk.rs

microsandbox_protocol/
bulk.rs
