rustferry_remote/
data_plane.rs

1//! Length-framed binary transport for large worker source and artifact payloads.
2
3use std::io::{Read, Write};
4
5use thiserror::Error;
6
7use crate::{MAX_SOURCE_BUNDLE_DESCRIPTOR_BYTES, protocol::MAX_EVENT_LINE_BYTES};
8
9/// Current schema for binary worker data-plane frame headers.
10pub const WORKER_DATA_PLANE_SCHEMA_VERSION: u16 = 1;
11/// Exact encoded size of one binary worker data-plane frame header.
12pub const WORKER_DATA_PLANE_HEADER_BYTES: usize = 24;
13/// Maximum build-request JSON size.
14pub const MAX_WORKER_DATA_PLANE_REQUEST_BYTES: u64 = 1024 * 1024;
15/// Maximum cancellation or acknowledgement JSON size.
16pub const MAX_WORKER_DATA_PLANE_CONTROL_BYTES: u64 = 64 * 1024;
17/// Maximum terminal result JSON size.
18pub const MAX_WORKER_DATA_PLANE_RESULT_BYTES: u64 = MAX_SOURCE_BUNDLE_DESCRIPTOR_BYTES;
19/// Maximum snapshot ZIP accepted from a client.
20pub const MAX_WORKER_DATA_PLANE_SOURCE_BYTES: u64 = 640 * 1024 * 1024;
21/// Maximum sealed physical-iPhone artifact returned by a worker.
22pub const MAX_WORKER_DATA_PLANE_ARTIFACT_BYTES: u64 = 2 * 1024 * 1024 * 1024;
23
24const WORKER_DATA_PLANE_MAGIC: [u8; 4] = *b"RFDP";
25
26/// Typed frame carried by the full-duplex worker data plane.
27#[derive(Clone, Copy, Debug, Eq, PartialEq)]
28#[repr(u16)]
29pub enum WorkerDataPlaneFrameKind {
30    /// Declarative [`crate::IosDeviceBuildRequest`] JSON.
31    BuildRequest = 1,
32    /// Strict [`crate::SourceBundleDescriptor`] JSON.
33    SourceDescriptor = 2,
34    /// Raw deterministic source ZIP bytes.
35    SourceArchive = 3,
36    /// In-band cancellation request JSON.
37    Cancel = 4,
38    /// Worker job acknowledgement JSON.
39    JobAccepted = 5,
40    /// One [`crate::RemoteBuildEvent`] JSON object.
41    Event = 6,
42    /// Strict artifact descriptor or manifest JSON.
43    ArtifactDescriptor = 7,
44    /// Raw artifact bytes bound by the preceding descriptor.
45    Artifact = 8,
46    /// Terminal build and cleanup result JSON.
47    Complete = 9,
48    /// Stable secret-free error JSON.
49    Error = 10,
50    /// Client proof that the offered artifact was verified and published.
51    ArtifactReceipt = 11,
52}
53
54impl WorkerDataPlaneFrameKind {
55    /// Maximum payload bytes accepted for this frame kind.
56    #[must_use]
57    pub const fn maximum_payload_bytes(self) -> u64 {
58        match self {
59            Self::BuildRequest => MAX_WORKER_DATA_PLANE_REQUEST_BYTES,
60            Self::SourceDescriptor | Self::ArtifactDescriptor => MAX_SOURCE_BUNDLE_DESCRIPTOR_BYTES,
61            Self::SourceArchive => MAX_WORKER_DATA_PLANE_SOURCE_BYTES,
62            Self::Cancel | Self::JobAccepted | Self::Error | Self::ArtifactReceipt => {
63                MAX_WORKER_DATA_PLANE_CONTROL_BYTES
64            }
65            Self::Event => MAX_EVENT_LINE_BYTES as u64,
66            Self::Artifact => MAX_WORKER_DATA_PLANE_ARTIFACT_BYTES,
67            Self::Complete => MAX_WORKER_DATA_PLANE_RESULT_BYTES,
68        }
69    }
70
71    fn from_wire(value: u16) -> Result<Self, WorkerDataPlaneFrameError> {
72        match value {
73            1 => Ok(Self::BuildRequest),
74            2 => Ok(Self::SourceDescriptor),
75            3 => Ok(Self::SourceArchive),
76            4 => Ok(Self::Cancel),
77            5 => Ok(Self::JobAccepted),
78            6 => Ok(Self::Event),
79            7 => Ok(Self::ArtifactDescriptor),
80            8 => Ok(Self::Artifact),
81            9 => Ok(Self::Complete),
82            10 => Ok(Self::Error),
83            11 => Ok(Self::ArtifactReceipt),
84            _ => Err(WorkerDataPlaneFrameError::UnknownFrameKind),
85        }
86    }
87}
88
89/// Validated header for one length-framed worker payload.
90#[derive(Clone, Copy, Debug, Eq, PartialEq)]
91pub struct WorkerDataPlaneFrameHeader {
92    kind: WorkerDataPlaneFrameKind,
93    sequence: u64,
94    payload_bytes: u64,
95}
96
97/// Per-direction monotonic frame-sequence validator.
98#[derive(Clone, Copy, Debug, Eq, PartialEq)]
99pub struct WorkerDataPlaneSequence {
100    next: Option<u64>,
101}
102
103impl WorkerDataPlaneSequence {
104    /// Start a stream at sequence zero.
105    #[must_use]
106    pub const fn new() -> Self {
107        Self { next: Some(0) }
108    }
109
110    /// Accept exactly the next header in this direction.
111    ///
112    /// # Errors
113    ///
114    /// Rejects gaps, duplicates, replay, and frames after sequence exhaustion.
115    pub fn accept(
116        &mut self,
117        header: WorkerDataPlaneFrameHeader,
118    ) -> Result<(), WorkerDataPlaneFrameError> {
119        let Some(expected) = self.next else {
120            return Err(WorkerDataPlaneFrameError::SequenceExhausted);
121        };
122        if header.sequence != expected {
123            return Err(WorkerDataPlaneFrameError::UnexpectedSequence {
124                expected,
125                received: header.sequence,
126            });
127        }
128        self.next = expected.checked_add(1);
129        Ok(())
130    }
131}
132
133impl Default for WorkerDataPlaneSequence {
134    fn default() -> Self {
135        Self::new()
136    }
137}
138
139impl WorkerDataPlaneFrameHeader {
140    /// Construct one bounded frame header.
141    ///
142    /// # Errors
143    ///
144    /// Rejects payload sizes above the fixed limit for `kind`.
145    pub fn new(
146        kind: WorkerDataPlaneFrameKind,
147        sequence: u64,
148        payload_bytes: u64,
149    ) -> Result<Self, WorkerDataPlaneFrameError> {
150        let maximum = kind.maximum_payload_bytes();
151        if payload_bytes > maximum {
152            return Err(WorkerDataPlaneFrameError::PayloadTooLarge {
153                kind,
154                payload_bytes,
155                maximum,
156            });
157        }
158        Ok(Self {
159            kind,
160            sequence,
161            payload_bytes,
162        })
163    }
164
165    /// Typed payload interpretation.
166    #[must_use]
167    pub const fn kind(self) -> WorkerDataPlaneFrameKind {
168        self.kind
169    }
170
171    /// Monotonic sequence within one stream direction.
172    #[must_use]
173    pub const fn sequence(self) -> u64 {
174        self.sequence
175    }
176
177    /// Exact following payload size.
178    #[must_use]
179    pub const fn payload_bytes(self) -> u64 {
180        self.payload_bytes
181    }
182}
183
184/// Stable binary-frame parsing or serialization failure.
185#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
186pub enum WorkerDataPlaneFrameError {
187    /// Reading or writing the stream failed.
188    #[error("worker data-plane I/O failed")]
189    Io,
190    /// No frame header was supplied.
191    #[error("worker data-plane stream is empty")]
192    EmptyInput,
193    /// Input ended inside a frame header.
194    #[error("worker data-plane frame header is truncated")]
195    TruncatedHeader,
196    /// Header magic does not identify the `RustFerry` data plane.
197    #[error("worker data-plane frame magic is invalid")]
198    InvalidMagic,
199    /// Header schema is not supported.
200    #[error("worker data-plane frame schema is unsupported")]
201    UnsupportedSchemaVersion {
202        /// Supported header schema.
203        supported: u16,
204        /// Received header schema.
205        received: u16,
206    },
207    /// Frame kind has no defined v1 meaning.
208    #[error("worker data-plane frame kind is unknown")]
209    UnknownFrameKind,
210    /// Declared payload exceeds the fixed bound for its kind.
211    #[error("worker data-plane frame payload exceeds its limit")]
212    PayloadTooLarge {
213        /// Frame interpretation.
214        kind: WorkerDataPlaneFrameKind,
215        /// Declared payload size.
216        payload_bytes: u64,
217        /// Maximum accepted size.
218        maximum: u64,
219    },
220    /// Input ended before the declared payload length.
221    #[error("worker data-plane frame payload is truncated")]
222    TruncatedPayload,
223    /// A small payload cannot be represented in local address space.
224    #[error("worker data-plane frame payload cannot be represented")]
225    PayloadSizeUnsupported,
226    /// A raw source or artifact payload must be copied into a bounded sink.
227    #[error("worker data-plane raw payload requires streaming")]
228    StreamingRequired {
229        /// Raw frame interpretation.
230        kind: WorkerDataPlaneFrameKind,
231    },
232    /// A frame was replayed, skipped, or reordered within its direction.
233    #[error("worker data-plane frame sequence is invalid")]
234    UnexpectedSequence {
235        /// Exact next sequence required.
236        expected: u64,
237        /// Sequence received from the peer.
238        received: u64,
239    },
240    /// No frame may follow sequence `u64::MAX`.
241    #[error("worker data-plane frame sequence is exhausted")]
242    SequenceExhausted,
243}
244
245/// Encode one validated header without buffering its payload.
246///
247/// # Errors
248///
249/// Returns a stable I/O error if the complete header cannot be written.
250pub fn write_worker_data_plane_header(
251    writer: &mut impl Write,
252    header: WorkerDataPlaneFrameHeader,
253) -> Result<(), WorkerDataPlaneFrameError> {
254    let mut encoded = [0_u8; WORKER_DATA_PLANE_HEADER_BYTES];
255    encoded[..4].copy_from_slice(&WORKER_DATA_PLANE_MAGIC);
256    encoded[4..6].copy_from_slice(&WORKER_DATA_PLANE_SCHEMA_VERSION.to_be_bytes());
257    encoded[6..8].copy_from_slice(&(header.kind as u16).to_be_bytes());
258    encoded[8..16].copy_from_slice(&header.sequence.to_be_bytes());
259    encoded[16..24].copy_from_slice(&header.payload_bytes.to_be_bytes());
260    writer
261        .write_all(&encoded)
262        .map_err(|_| WorkerDataPlaneFrameError::Io)
263}
264
265/// Decode and validate one fixed-size frame header without reading its payload.
266///
267/// # Errors
268///
269/// Distinguishes empty input, truncation, schema/kind mismatch, and an
270/// over-limit declaration before payload allocation or I/O.
271pub fn read_worker_data_plane_header(
272    reader: &mut impl Read,
273) -> Result<WorkerDataPlaneFrameHeader, WorkerDataPlaneFrameError> {
274    let mut encoded = [0_u8; WORKER_DATA_PLANE_HEADER_BYTES];
275    read_header_exact(reader, &mut encoded)?;
276    if encoded[..4] != WORKER_DATA_PLANE_MAGIC {
277        return Err(WorkerDataPlaneFrameError::InvalidMagic);
278    }
279    let schema_version = u16::from_be_bytes([encoded[4], encoded[5]]);
280    if schema_version != WORKER_DATA_PLANE_SCHEMA_VERSION {
281        return Err(WorkerDataPlaneFrameError::UnsupportedSchemaVersion {
282            supported: WORKER_DATA_PLANE_SCHEMA_VERSION,
283            received: schema_version,
284        });
285    }
286    let kind = WorkerDataPlaneFrameKind::from_wire(u16::from_be_bytes([encoded[6], encoded[7]]))?;
287    let sequence = u64::from_be_bytes([
288        encoded[8],
289        encoded[9],
290        encoded[10],
291        encoded[11],
292        encoded[12],
293        encoded[13],
294        encoded[14],
295        encoded[15],
296    ]);
297    let payload_bytes = u64::from_be_bytes([
298        encoded[16],
299        encoded[17],
300        encoded[18],
301        encoded[19],
302        encoded[20],
303        encoded[21],
304        encoded[22],
305        encoded[23],
306    ]);
307    WorkerDataPlaneFrameHeader::new(kind, sequence, payload_bytes)
308}
309
310/// Write one bounded in-memory frame.
311///
312/// # Errors
313///
314/// Rejects an oversized payload before writing and returns a stable I/O error
315/// if the header or complete payload cannot be written.
316pub fn write_worker_data_plane_frame(
317    writer: &mut impl Write,
318    kind: WorkerDataPlaneFrameKind,
319    sequence: u64,
320    payload: &[u8],
321) -> Result<(), WorkerDataPlaneFrameError> {
322    let payload_bytes = u64::try_from(payload.len())
323        .map_err(|_| WorkerDataPlaneFrameError::PayloadSizeUnsupported)?;
324    let header = WorkerDataPlaneFrameHeader::new(kind, sequence, payload_bytes)?;
325    write_worker_data_plane_header(writer, header)?;
326    writer
327        .write_all(payload)
328        .map_err(|_| WorkerDataPlaneFrameError::Io)
329}
330
331/// Write one header followed by an exact-length streamed payload.
332///
333/// The caller must separately bind raw payload bytes to a trusted descriptor
334/// and verify its digest. This function guarantees framing and constant memory.
335///
336/// # Errors
337///
338/// Rejects an over-limit declaration, a short source, or an I/O failure.
339pub fn write_worker_data_plane_stream(
340    writer: &mut impl Write,
341    kind: WorkerDataPlaneFrameKind,
342    sequence: u64,
343    reader: &mut impl Read,
344    payload_bytes: u64,
345) -> Result<u64, WorkerDataPlaneFrameError> {
346    let header = WorkerDataPlaneFrameHeader::new(kind, sequence, payload_bytes)?;
347    write_worker_data_plane_header(writer, header)?;
348    copy_worker_data_plane_payload(reader, writer, header)
349}
350
351/// Read one bounded in-memory payload after its validated header.
352///
353/// Raw source and artifact frames should instead use
354/// [`copy_worker_data_plane_payload`] to avoid allocating their declared size.
355///
356/// # Errors
357///
358/// Rejects sizes unavailable in local address space and distinguishes a short
359/// payload from another I/O failure.
360pub fn read_worker_data_plane_payload(
361    reader: &mut impl Read,
362    header: WorkerDataPlaneFrameHeader,
363) -> Result<Vec<u8>, WorkerDataPlaneFrameError> {
364    if matches!(
365        header.kind,
366        WorkerDataPlaneFrameKind::SourceArchive | WorkerDataPlaneFrameKind::Artifact
367    ) {
368        return Err(WorkerDataPlaneFrameError::StreamingRequired { kind: header.kind });
369    }
370    let length = usize::try_from(header.payload_bytes)
371        .map_err(|_| WorkerDataPlaneFrameError::PayloadSizeUnsupported)?;
372    let mut payload = vec![0_u8; length];
373    read_payload_exact(reader, &mut payload)?;
374    Ok(payload)
375}
376
377/// Stream exactly one validated payload into a sink with constant memory.
378///
379/// # Errors
380///
381/// Returns [`WorkerDataPlaneFrameError::TruncatedPayload`] if input ends before
382/// the declared length, or a stable I/O error if reading or writing otherwise
383/// fails.
384pub fn copy_worker_data_plane_payload(
385    reader: &mut impl Read,
386    writer: &mut impl Write,
387    header: WorkerDataPlaneFrameHeader,
388) -> Result<u64, WorkerDataPlaneFrameError> {
389    let mut remaining = header.payload_bytes;
390    let mut copied = 0_u64;
391    let mut buffer = [0_u8; 16 * 1024];
392    while remaining != 0 {
393        let chunk = usize::try_from(remaining.min(buffer.len() as u64))
394            .map_err(|_| WorkerDataPlaneFrameError::PayloadSizeUnsupported)?;
395        let count = reader
396            .read(&mut buffer[..chunk])
397            .map_err(|_| WorkerDataPlaneFrameError::Io)?;
398        if count == 0 {
399            return Err(WorkerDataPlaneFrameError::TruncatedPayload);
400        }
401        writer
402            .write_all(&buffer[..count])
403            .map_err(|_| WorkerDataPlaneFrameError::Io)?;
404        let count = count as u64;
405        remaining -= count;
406        copied += count;
407    }
408    Ok(copied)
409}
410
411fn read_header_exact(
412    reader: &mut impl Read,
413    output: &mut [u8],
414) -> Result<(), WorkerDataPlaneFrameError> {
415    let mut offset = 0;
416    while offset < output.len() {
417        let count = reader
418            .read(&mut output[offset..])
419            .map_err(|_| WorkerDataPlaneFrameError::Io)?;
420        if count == 0 {
421            return Err(if offset == 0 {
422                WorkerDataPlaneFrameError::EmptyInput
423            } else {
424                WorkerDataPlaneFrameError::TruncatedHeader
425            });
426        }
427        offset += count;
428    }
429    Ok(())
430}
431
432fn read_payload_exact(
433    reader: &mut impl Read,
434    output: &mut [u8],
435) -> Result<(), WorkerDataPlaneFrameError> {
436    let mut offset = 0;
437    while offset < output.len() {
438        let count = reader
439            .read(&mut output[offset..])
440            .map_err(|_| WorkerDataPlaneFrameError::Io)?;
441        if count == 0 {
442            return Err(WorkerDataPlaneFrameError::TruncatedPayload);
443        }
444        offset += count;
445    }
446    Ok(())
447}
448
449#[cfg(test)]
450mod tests {
451    use std::io::Cursor;
452
453    use super::*;
454
455    #[test]
456    fn concatenated_frames_round_trip_without_consuming_each_other() {
457        let mut bytes = Vec::new();
458        write_worker_data_plane_frame(
459            &mut bytes,
460            WorkerDataPlaneFrameKind::BuildRequest,
461            0,
462            br#"{"request":"one"}"#,
463        )
464        .expect("request frame");
465        write_worker_data_plane_frame(
466            &mut bytes,
467            WorkerDataPlaneFrameKind::Cancel,
468            1,
469            br#"{"reason":"user_requested"}"#,
470        )
471        .expect("cancel frame");
472
473        let mut input = Cursor::new(bytes);
474        let first = read_worker_data_plane_header(&mut input).expect("first header");
475        assert_eq!(first.kind(), WorkerDataPlaneFrameKind::BuildRequest);
476        assert_eq!(first.sequence(), 0);
477        assert_eq!(
478            read_worker_data_plane_payload(&mut input, first).expect("first payload"),
479            br#"{"request":"one"}"#
480        );
481        let second = read_worker_data_plane_header(&mut input).expect("second header");
482        assert_eq!(second.kind(), WorkerDataPlaneFrameKind::Cancel);
483        assert_eq!(second.sequence(), 1);
484        assert_eq!(
485            read_worker_data_plane_payload(&mut input, second).expect("second payload"),
486            br#"{"reason":"user_requested"}"#
487        );
488        assert_eq!(
489            read_worker_data_plane_header(&mut input),
490            Err(WorkerDataPlaneFrameError::EmptyInput)
491        );
492    }
493
494    #[test]
495    fn malformed_headers_are_rejected_before_payload_reads() {
496        let valid = encoded_header(WorkerDataPlaneFrameKind::Event, 7, 0);
497        let mut wrong_magic = valid;
498        wrong_magic[0] = b'X';
499        assert_eq!(
500            read_worker_data_plane_header(&mut Cursor::new(wrong_magic)),
501            Err(WorkerDataPlaneFrameError::InvalidMagic)
502        );
503
504        let mut wrong_schema = valid;
505        wrong_schema[4..6].copy_from_slice(&2_u16.to_be_bytes());
506        assert_eq!(
507            read_worker_data_plane_header(&mut Cursor::new(wrong_schema)),
508            Err(WorkerDataPlaneFrameError::UnsupportedSchemaVersion {
509                supported: WORKER_DATA_PLANE_SCHEMA_VERSION,
510                received: 2,
511            })
512        );
513
514        let mut unknown_kind = valid;
515        unknown_kind[6..8].copy_from_slice(&u16::MAX.to_be_bytes());
516        assert_eq!(
517            read_worker_data_plane_header(&mut Cursor::new(unknown_kind)),
518            Err(WorkerDataPlaneFrameError::UnknownFrameKind)
519        );
520    }
521
522    #[test]
523    fn empty_truncated_header_and_payload_are_distinct() {
524        assert_eq!(
525            read_worker_data_plane_header(&mut Cursor::new(Vec::<u8>::new())),
526            Err(WorkerDataPlaneFrameError::EmptyInput)
527        );
528        assert_eq!(
529            read_worker_data_plane_header(&mut Cursor::new(vec![0_u8; 12])),
530            Err(WorkerDataPlaneFrameError::TruncatedHeader)
531        );
532
533        let header = WorkerDataPlaneFrameHeader::new(WorkerDataPlaneFrameKind::Event, 0, 4)
534            .expect("bounded header");
535        assert_eq!(
536            read_worker_data_plane_payload(&mut Cursor::new(b"abc"), header),
537            Err(WorkerDataPlaneFrameError::TruncatedPayload)
538        );
539    }
540
541    #[test]
542    fn oversized_declaration_is_rejected_from_header_alone() {
543        let encoded = encoded_header(
544            WorkerDataPlaneFrameKind::SourceArchive,
545            0,
546            MAX_WORKER_DATA_PLANE_SOURCE_BYTES + 1,
547        );
548        assert_eq!(
549            read_worker_data_plane_header(&mut Cursor::new(encoded)),
550            Err(WorkerDataPlaneFrameError::PayloadTooLarge {
551                kind: WorkerDataPlaneFrameKind::SourceArchive,
552                payload_bytes: MAX_WORKER_DATA_PLANE_SOURCE_BYTES + 1,
553                maximum: MAX_WORKER_DATA_PLANE_SOURCE_BYTES,
554            })
555        );
556    }
557
558    #[test]
559    fn raw_payload_copy_is_exact_and_constant_memory() {
560        let payload = vec![0x5a; 3 * 64 * 1024 + 17];
561        let header = WorkerDataPlaneFrameHeader::new(
562            WorkerDataPlaneFrameKind::SourceArchive,
563            2,
564            payload.len() as u64,
565        )
566        .expect("bounded source frame");
567        let mut output = Vec::new();
568        assert_eq!(
569            copy_worker_data_plane_payload(&mut Cursor::new(&payload), &mut output, header)
570                .expect("streamed payload"),
571            payload.len() as u64
572        );
573        assert_eq!(output, payload);
574    }
575
576    #[test]
577    fn raw_payload_copy_reports_truncation() {
578        let header = WorkerDataPlaneFrameHeader::new(WorkerDataPlaneFrameKind::Artifact, 4, 4)
579            .expect("bounded artifact frame");
580        assert_eq!(
581            copy_worker_data_plane_payload(&mut Cursor::new(b"abc"), &mut Vec::new(), header),
582            Err(WorkerDataPlaneFrameError::TruncatedPayload)
583        );
584    }
585
586    #[test]
587    fn raw_payload_cannot_enter_the_in_memory_reader() {
588        for kind in [
589            WorkerDataPlaneFrameKind::SourceArchive,
590            WorkerDataPlaneFrameKind::Artifact,
591        ] {
592            let header = WorkerDataPlaneFrameHeader::new(kind, 0, 1).expect("bounded raw frame");
593            assert_eq!(
594                read_worker_data_plane_payload(&mut Cursor::new([0_u8]), header),
595                Err(WorkerDataPlaneFrameError::StreamingRequired { kind })
596            );
597        }
598    }
599
600    #[test]
601    fn sequence_rejects_gap_replay_and_exhaustion() {
602        let mut sequence = WorkerDataPlaneSequence::new();
603        let first = WorkerDataPlaneFrameHeader::new(WorkerDataPlaneFrameKind::BuildRequest, 0, 0)
604            .expect("first header");
605        sequence.accept(first).expect("first sequence");
606        let gap = WorkerDataPlaneFrameHeader::new(WorkerDataPlaneFrameKind::Cancel, 2, 0)
607            .expect("gap header");
608        assert_eq!(
609            sequence.accept(gap),
610            Err(WorkerDataPlaneFrameError::UnexpectedSequence {
611                expected: 1,
612                received: 2,
613            })
614        );
615        let replay = WorkerDataPlaneFrameHeader::new(WorkerDataPlaneFrameKind::Cancel, 0, 0)
616            .expect("replay header");
617        assert_eq!(
618            sequence.accept(replay),
619            Err(WorkerDataPlaneFrameError::UnexpectedSequence {
620                expected: 1,
621                received: 0,
622            })
623        );
624
625        let mut exhausted = WorkerDataPlaneSequence {
626            next: Some(u64::MAX),
627        };
628        let last = WorkerDataPlaneFrameHeader::new(WorkerDataPlaneFrameKind::Complete, u64::MAX, 0)
629            .expect("last header");
630        exhausted.accept(last).expect("last sequence");
631        assert_eq!(
632            exhausted.accept(last),
633            Err(WorkerDataPlaneFrameError::SequenceExhausted)
634        );
635    }
636
637    #[test]
638    fn streamed_frame_writes_exact_payload_without_buffering_it() {
639        let payload = vec![0x3c; 2 * 64 * 1024 + 9];
640        let mut encoded = Vec::new();
641        assert_eq!(
642            write_worker_data_plane_stream(
643                &mut encoded,
644                WorkerDataPlaneFrameKind::Artifact,
645                8,
646                &mut Cursor::new(&payload),
647                payload.len() as u64,
648            )
649            .expect("streamed frame"),
650            payload.len() as u64
651        );
652        let mut input = Cursor::new(encoded);
653        let header = read_worker_data_plane_header(&mut input).expect("stream header");
654        assert_eq!(header.kind(), WorkerDataPlaneFrameKind::Artifact);
655        assert_eq!(header.sequence(), 8);
656        let mut output = Vec::new();
657        copy_worker_data_plane_payload(&mut input, &mut output, header).expect("stream payload");
658        assert_eq!(output, payload);
659    }
660
661    fn encoded_header(
662        kind: WorkerDataPlaneFrameKind,
663        sequence: u64,
664        payload_bytes: u64,
665    ) -> [u8; WORKER_DATA_PLANE_HEADER_BYTES] {
666        let mut bytes = Vec::new();
667        let header = WorkerDataPlaneFrameHeader {
668            kind,
669            sequence,
670            payload_bytes,
671        };
672        write_worker_data_plane_header(&mut bytes, header).expect("encoded header");
673        bytes.try_into().expect("fixed header size")
674    }
675}
rustferry_remote/data_plane.rs

rustferry_remote/
data_plane.rs
