buntrail_services/git/proof_codec/
output.rs

1//! One status entry or original alias at a time; never an aggregate service JSON reply.
2use super::*;
3use crate::git::observation::ObservationRecord;
4const OUTPUT: &[u8; 8] = b"BTOUTP01";
5#[derive(Debug, Clone, PartialEq, Eq)]
6pub struct OwnedObservation {
7    pub id: [u8; 16],
8    pub worktree: WorktreeIdentity,
9    pub state: GitExpectedState,
10    pub alias_digest: [u8; 32],
11    pub retained_bytes: u64,
12}
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14#[repr(u8)]
15pub enum EntryGroup {
16    Staged = 1,
17    Unstaged = 2,
18    Untracked = 3,
19    Conflict = 4,
20}
21#[derive(Debug, Clone, PartialEq, Eq)]
22// One record is streamed at a time; boxing Alias would allocate for every original alias.
23#[allow(clippy::large_enum_variant)]
24pub enum ObservationEvent {
25    Entry { group: EntryGroup, entry: GitEntry },
26    Alias(ObservationRecord),
27}
28fn aliases_hash() -> Sha256 {
29    let mut hash = Sha256::new();
30    hash.update(b"buntrail.original-aliases.v1\0");
31    hash
32}
33fn checked_path(path: &TargetPath, limit: TargetPathLimit) -> io::Result<&[u8]> {
34    let b = path_bytes(path)?;
35    if b.len() > limit.0 as usize || b.first() != Some(&b'/') || b.contains(&0) {
36        return Err(malformed());
37    }
38    Ok(b)
39}
40pub struct OutputWriter<W> {
41    out: W,
42    limit: TargetPathLimit,
43    counts: [u64; 5],
44    phase: u8,
45    hash: Sha256,
46    expected: [u8; 32],
47    failed: bool,
48}
49impl<W: Write> OutputWriter<W> {
50    pub fn new(
51        mut out: W,
52        summary: &OwnedObservation,
53        limit: TargetPathLimit,
54    ) -> Result<Self, CoreError> {
55        (|| -> io::Result<()> {
56            out.write_all(OUTPUT)?;
57            out.write_all(&limit.0.to_be_bytes())?;
58            out.write_all(&summary.id)?;
59            out.write_all(&summary.retained_bytes.to_be_bytes())?;
60            bytes(
61                &mut out,
62                checked_path(&summary.worktree.canonical_path, limit)?,
63            )?;
64            bytes(&mut out, &summary.worktree.common_git_directory_identity)?;
65            match &summary.state.head {
66                GitHead::Branch { name, oid } => {
67                    tag(&mut out, 0)?;
68                    bytes(&mut out, name)?;
69                    bytes(&mut out, oid.as_bytes())?;
70                }
71                GitHead::Detached(oid) => {
72                    tag(&mut out, 1)?;
73                    bytes(&mut out, oid.as_bytes())?;
74                }
75                GitHead::Unborn(name) => {
76                    tag(&mut out, 2)?;
77                    bytes(&mut out, name)?;
78                }
79            }
80            out.write_all(&summary.state.index_fingerprint)?;
81            out.write_all(&summary.state.worktree_fingerprint)?;
82            out.write_all(&summary.alias_digest)
83        })()
84        .map_err(codec_error)?;
85        Ok(Self {
86            out,
87            limit,
88            counts: [0; 5],
89            phase: 0,
90            hash: aliases_hash(),
91            expected: summary.alias_digest,
92            failed: false,
93        })
94    }
95    pub fn event(&mut self, event: &ObservationEvent) -> Result<(), CoreError> {
96        if self.failed {
97            return Err(invalid());
98        }
99        self.failed = true;
100        let group = match event {
101            ObservationEvent::Entry { group, .. } => *group as u8,
102            ObservationEvent::Alias(_) => 5,
103        };
104        if group < self.phase {
105            return Err(invalid());
106        }
107        self.phase = group;
108        self.counts[group as usize - 1] = self.counts[group as usize - 1]
109            .checked_add(1)
110            .ok_or_else(invalid)?;
111        (|| -> io::Result<()> {
112            tag(&mut self.out, group)?;
113            match event {
114                ObservationEvent::Entry { entry, .. } => {
115                    bytes(&mut self.out, checked_path(&entry.path, self.limit)?)?;
116                    match &entry.previous_path {
117                        None => tag(&mut self.out, 0)?,
118                        Some(path) => {
119                            tag(&mut self.out, 1)?;
120                            bytes(&mut self.out, checked_path(path, self.limit)?)?;
121                        }
122                    }
123                    bytes(&mut self.out, entry.status.as_bytes())?;
124                }
125                ObservationEvent::Alias(record) => {
126                    if record.membership.is_none() {
127                        return Err(malformed());
128                    }
129                    checked_path(&record.path, self.limit)?;
130                    if let Some(path) = &record.resolved {
131                        checked_path(path, self.limit)?;
132                    }
133                    let digest = record.file_reference.ok_or_else(malformed)?;
134                    self.out
135                        .write_all(&record.encode().map_err(|_| malformed())?)?;
136                    self.hash.update(digest);
137                }
138            }
139            Ok(())
140        })()
141        .map_err(codec_error)?;
142        self.failed = false;
143        Ok(())
144    }
145    pub fn finish(mut self) -> Result<W, CoreError> {
146        if self.failed || <[u8; 32]>::from(self.hash.finalize()) != self.expected {
147            return Err(invalid());
148        }
149        tag(&mut self.out, 0).map_err(codec_error)?;
150        for count in self.counts {
151            self.out.write_all(&count.to_be_bytes()).map_err(io_error)?;
152        }
153        self.out.write_all(&self.expected).map_err(io_error)?;
154        Ok(self.out)
155    }
156}
157pub struct OutputReader<R> {
158    input: Option<Input<R>>,
159    summary: OwnedObservation,
160    limit: TargetPathLimit,
161    counts: [u64; 5],
162    phase: u8,
163    hash: Sha256,
164    complete: bool,
165    wire_digest: Option<[u8; 32]>,
166}
167impl<R: Read> OutputReader<R> {
168    pub fn new(reader: R, length: u64, target: &TargetIdentity) -> Result<Self, CoreError> {
169        let mut input = Input::new(reader, length);
170        let (summary, limit) = (|| -> Result<_, CoreError> {
171            if &input.fixed::<8>().map_err(codec_error)? != OUTPUT {
172                return Err(invalid());
173            }
174            let limit =
175                TargetPathLimit::declared(u32::from_be_bytes(input.fixed().map_err(codec_error)?))?;
176            let id = input.fixed().map_err(codec_error)?;
177            let retained_bytes = input.number().map_err(codec_error)?;
178            let worktree = WorktreeIdentity {
179                target: target.clone(),
180                canonical_path: TargetPath::posix(
181                    input.bounded(limit.0 as u64).map_err(codec_error)?,
182                )?,
183                common_git_directory_identity: input.bounded(SCALAR).map_err(codec_error)?,
184            };
185            let head = match input.tag().map_err(codec_error)? {
186                0 => GitHead::Branch {
187                    name: input.bounded(SCALAR).map_err(codec_error)?,
188                    oid: input.string().map_err(codec_error)?,
189                },
190                1 => GitHead::Detached(input.string().map_err(codec_error)?),
191                2 => GitHead::Unborn(input.bounded(SCALAR).map_err(codec_error)?),
192                _ => return Err(invalid()),
193            };
194            let state = GitExpectedState {
195                head,
196                index_fingerprint: input.fixed().map_err(codec_error)?,
197                worktree_fingerprint: input.fixed().map_err(codec_error)?,
198            };
199            let alias_digest = input.fixed().map_err(codec_error)?;
200            Ok((
201                OwnedObservation {
202                    id,
203                    worktree,
204                    state,
205                    alias_digest,
206                    retained_bytes,
207                },
208                limit,
209            ))
210        })()?;
211        Ok(Self {
212            input: Some(input),
213            summary,
214            limit,
215            counts: [0; 5],
216            phase: 0,
217            hash: aliases_hash(),
218            complete: false,
219            wire_digest: None,
220        })
221    }
222    /// Events are provisional until EOF, trailer and the transport's whole-body seal validate.
223    pub fn next_event(&mut self) -> Result<Option<ObservationEvent>, CoreError> {
224        if self.complete {
225            return Ok(None);
226        }
227        let result = self.event();
228        if result.is_err() {
229            self.input = None;
230        }
231        result
232    }
233    fn event(&mut self) -> Result<Option<ObservationEvent>, CoreError> {
234        let input = self.input.as_mut().ok_or_else(invalid)?;
235        let kind = input.tag().map_err(codec_error)?;
236        if kind == 0 {
237            for count in self.counts {
238                if input.number().map_err(codec_error)? != count {
239                    return Err(invalid());
240                }
241            }
242            let digest = input.fixed::<32>().map_err(codec_error)?;
243            if digest != self.summary.alias_digest
244                || digest != <[u8; 32]>::from(self.hash.clone().finalize())
245            {
246                return Err(invalid());
247            }
248            self.wire_digest = Some(self.input.take().unwrap().finish().map_err(codec_error)?);
249            self.complete = true;
250            return Ok(None);
251        }
252        if kind < self.phase || kind > 5 {
253            return Err(invalid());
254        }
255        self.phase = kind;
256        self.counts[kind as usize - 1] = self.counts[kind as usize - 1]
257            .checked_add(1)
258            .ok_or_else(invalid)?;
259        if kind == 5 {
260            let record = ObservationRecord::read_with_path_limit(input, self.limit.0 as usize)?
261                .ok_or_else(invalid)?;
262            if record.membership.is_none() {
263                return Err(invalid());
264            }
265            self.hash.update(record.file_reference.ok_or_else(invalid)?);
266            return Ok(Some(ObservationEvent::Alias(record)));
267        }
268        let path = TargetPath::posix(input.bounded(self.limit.0 as u64).map_err(codec_error)?)?;
269        let previous_path = match input.tag().map_err(codec_error)? {
270            0 => None,
271            1 => Some(TargetPath::posix(
272                input.bounded(self.limit.0 as u64).map_err(codec_error)?,
273            )?),
274            _ => return Err(invalid()),
275        };
276        let status = input.string().map_err(codec_error)?;
277        let group = match kind {
278            1 => EntryGroup::Staged,
279            2 => EntryGroup::Unstaged,
280            3 => EntryGroup::Untracked,
281            4 => EntryGroup::Conflict,
282            _ => return Err(invalid()),
283        };
284        Ok(Some(ObservationEvent::Entry {
285            group,
286            entry: GitEntry {
287                path,
288                previous_path,
289                status,
290            },
291        }))
292    }
293    /// Provisional identity for owner-scoped cleanup only; no observation authority before finish.
294    pub fn observation_id(&self) -> [u8; 16] {
295        self.summary.id
296    }
297    /// Compare with the authenticated transport's end seal before using the summary.
298    pub fn wire_digest(&self) -> Option<[u8; 32]> {
299        self.wire_digest
300    }
301    pub fn finish(self) -> Result<OwnedObservation, CoreError> {
302        if !self.complete {
303            return Err(invalid());
304        }
305        Ok(self.summary)
306    }
307}
308
309#[cfg(test)]
310mod tests {
311    use super::*;
312    use crate::git::observation::ObservationMembership;
313    fn fixture() -> (OwnedObservation, ObservationRecord) {
314        let r = super::super::tests::original();
315        let record = ObservationRecord {
316            path: TargetPath::PosixBytes([b"/".as_slice(), &vec![b'x'; 2999]].concat()),
317            structural_token: [1; 32],
318            resolved: None,
319            regular: None,
320            membership: Some(ObservationMembership {
321                requested_repository: Some([2; 32]),
322                resolved_repository: None,
323                affected: true,
324            }),
325            file_reference: Some([3; 32]),
326        };
327        let mut hash = Sha256::new();
328        hash.update(b"buntrail.original-aliases.v1\0");
329        hash.update([3; 32]);
330        (
331            OwnedObservation {
332                id: [4; 16],
333                worktree: r.worktree,
334                state: r.expected,
335                alias_digest: hash.finalize().into(),
336                retained_bytes: 999,
337            },
338            record,
339        )
340    }
341    #[test]
342    fn output_streams_all_status_groups_and_linux_paths_without_host_path_limit() {
343        let (summary, record) = fixture();
344        let mut bytes = Vec::new();
345        let mut writer = OutputWriter::new(
346            &mut bytes,
347            &summary,
348            TargetPathLimit::declared(4096).unwrap(),
349        )
350        .expect("owned output must stream");
351        for group in [
352            EntryGroup::Staged,
353            EntryGroup::Unstaged,
354            EntryGroup::Untracked,
355            EntryGroup::Conflict,
356        ] {
357            writer
358                .event(&ObservationEvent::Entry {
359                    group,
360                    entry: GitEntry {
361                        path: record.path.clone(),
362                        previous_path: Some(TargetPath::PosixBytes(b"/old\xff".to_vec())),
363                        status: "R ".into(),
364                    },
365                })
366                .unwrap();
367        }
368        writer
369            .event(&ObservationEvent::Alias(record.clone()))
370            .unwrap();
371        writer.finish().unwrap();
372        let mut reader = OutputReader::new(
373            super::super::tests::Chunks(&bytes, 7),
374            bytes.len() as u64,
375            &summary.worktree.target,
376        )
377        .unwrap();
378        for group in [
379            EntryGroup::Staged,
380            EntryGroup::Unstaged,
381            EntryGroup::Untracked,
382            EntryGroup::Conflict,
383        ] {
384            assert!(
385                matches!(reader.next_event().unwrap(),Some(ObservationEvent::Entry{group:actual,entry}) if actual==group&&entry.path==record.path)
386            );
387        }
388        assert_eq!(
389            reader.next_event().unwrap(),
390            Some(ObservationEvent::Alias(record))
391        );
392        assert_eq!(reader.next_event().unwrap(), None);
393        assert_eq!(reader.finish().unwrap(), summary);
394        bytes[8..12].copy_from_slice(&1024u32.to_be_bytes());
395        let mut reader = OutputReader::new(
396            bytes.as_slice(),
397            bytes.len() as u64,
398            &summary.worktree.target,
399        )
400        .unwrap();
401        assert!(reader.next_event().is_err());
402    }
403    #[test]
404    fn output_exceeds_legacy_32_mib_without_an_aggregate_buffer() {
405        use std::io::{Seek, SeekFrom};
406        let (summary, record) = fixture();
407        let mut file = tempfile::tempfile().unwrap();
408        let mut writer = OutputWriter::new(
409            &mut file,
410            &summary,
411            TargetPathLimit::declared(4096).unwrap(),
412        )
413        .unwrap();
414        let event = ObservationEvent::Entry {
415            group: EntryGroup::Untracked,
416            entry: GitEntry {
417                path: record.path.clone(),
418                previous_path: None,
419                status: "??".into(),
420            },
421        };
422        for _ in 0..12000 {
423            writer.event(&event).unwrap();
424        }
425        writer.event(&ObservationEvent::Alias(record)).unwrap();
426        writer.finish().unwrap();
427        let length = file.stream_position().unwrap();
428        assert!(length > 32 * 1024 * 1024);
429        file.seek(SeekFrom::Start(0)).unwrap();
430        let mut reader = OutputReader::new(file, length, &summary.worktree.target).unwrap();
431        let mut count = 0;
432        while reader.next_event().unwrap().is_some() {
433            count += 1;
434        }
435        assert_eq!(count, 12001);
436        assert_eq!(reader.finish().unwrap(), summary);
437        eprintln!(
438            "synthetic owned output: {length} bytes, {count} events; no Git scalability claim"
439        );
440    }
441    #[test]
442    fn output_rejects_bad_trailer_truncation_order_type_and_unsupported_path_bound() {
443        let (summary, record) = fixture();
444        let mut bytes = Vec::new();
445        let mut writer = OutputWriter::new(
446            &mut bytes,
447            &summary,
448            TargetPathLimit::declared(4096).unwrap(),
449        )
450        .unwrap();
451        writer
452            .event(&ObservationEvent::Alias(record.clone()))
453            .unwrap();
454        writer.finish().unwrap();
455        let decode = |b: &[u8]| -> Result<OwnedObservation, CoreError> {
456            let mut r = OutputReader::new(
457                super::super::tests::Chunks(b, 1),
458                b.len() as u64,
459                &summary.worktree.target,
460            )?;
461            while r.next_event()?.is_some() {}
462            r.finish()
463        };
464        for n in [0, 8, 12, 28, bytes.len() - 1] {
465            assert!(decode(&bytes[..n]).is_err());
466        }
467        let mut changed = bytes.clone();
468        let end = changed.len() - 1;
469        changed[end] ^= 1;
470        assert!(decode(&changed).is_err());
471        let mut changed = bytes.clone();
472        changed[8..12].copy_from_slice(&u32::MAX.to_be_bytes());
473        assert!(decode(&changed).is_err());
474        let mut changed = bytes.clone();
475        changed.push(0);
476        assert!(decode(&changed).is_err());
477        let mut writer = OutputWriter::new(
478            Vec::new(),
479            &summary,
480            TargetPathLimit::declared(4096).unwrap(),
481        )
482        .unwrap();
483        writer
484            .event(&ObservationEvent::Alias(record.clone()))
485            .unwrap();
486        assert!(
487            writer
488                .event(&ObservationEvent::Entry {
489                    group: EntryGroup::Staged,
490                    entry: GitEntry {
491                        path: record.path.clone(),
492                        previous_path: None,
493                        status: "M".into()
494                    }
495                })
496                .is_err()
497        );
498        assert!(writer.finish().is_err());
499        let mut writer = OutputWriter::new(
500            Vec::new(),
501            &summary,
502            TargetPathLimit::declared(4096).unwrap(),
503        )
504        .unwrap();
505        let mut wrong = record;
506        wrong.membership = None;
507        assert!(writer.event(&ObservationEvent::Alias(wrong)).is_err());
508        assert!(
509            OutputReader::new(
510                bytes.as_slice(),
511                bytes.len() as u64,
512                &summary.worktree.target
513            )
514            .unwrap()
515            .finish()
516            .is_err(),
517            "partial output granted summary"
518        );
519    }
520    #[test]
521    fn owned_output_uses_proof_store_bytes_and_releases_its_input_slot() {
522        let dir = tempfile::tempdir().unwrap();
523        let local =
524            crate::connections::local::LocalConnection::open(dir.path().join("state")).unwrap();
525        let lease = WorkspaceLeaseID([92; 16]);
526        local.register(WorkspaceID([92; 16]), lease).unwrap();
527        let owner = local.observation_owner(lease).unwrap();
528        let (mut summary, mut record) = fixture();
529        summary.worktree.target = local.identity.clone();
530        record.path = TargetPath::PosixBytes(b"/alias".to_vec());
531        let spool = local.git.observation_input(&owner).unwrap();
532        let mut output = OutputWriter::new(
533            std::io::BufWriter::with_capacity(65536, spool),
534            &summary,
535            TargetPathLimit::declared(libc::PATH_MAX as u32).unwrap(),
536        )
537        .unwrap();
538        output.event(&ObservationEvent::Alias(record)).unwrap();
539        let buffered = output.finish().unwrap();
540        let input = buffered
541            .into_inner()
542            .map_err(|e| e.into_error())
543            .unwrap()
544            .seal()
545            .unwrap();
546        assert_eq!(
547            local.git.observation_storage_stats().unwrap().bytes,
548            input.len()
549        );
550        assert!(local.git.observation_input(&owner).is_err());
551        let mut reader =
552            OutputReader::new(input.reader().unwrap(), input.len(), &local.identity).unwrap();
553        while reader.next_event().unwrap().is_some() {}
554        assert_eq!(reader.wire_digest(), Some(input.digest));
555        assert_eq!(reader.finish().unwrap(), summary);
556        drop(input);
557        assert_eq!(local.git.observation_storage_stats().unwrap().bytes, 0);
558        local.git.observation_input(&owner).unwrap();
559    }
560}
buntrail_services/git/proof_codec/output.rs

buntrail_services/git/proof_codec/
output.rs
