1use std::collections::{BTreeMap, BTreeSet};
12use std::fmt;
13use std::fs::{File, Metadata};
14use std::io::{self, Read};
15use std::path::{Path, PathBuf};
16use std::time::SystemTime;
17
18pub use crate::hashline::oracle::{normalize_for_tag, tag_for};
19
20pub const MAX_FILE_READ_BYTES: u64 = 64 * 1024 * 1024;
25
26#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
28pub enum TerminatorKind {
29 Lf,
31 CrLf,
34 None,
36}
37
38impl TerminatorKind {
39 pub const LF: Self = Self::Lf;
42 pub const CRLF: Self = Self::CrLf;
43}
44
45pub type Terminator = TerminatorKind;
46
47impl fmt::Display for TerminatorKind {
48 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
49 formatter.write_str(match self {
50 Self::Lf => "LF",
51 Self::CrLf => "CRLF",
52 Self::None => "none",
53 })
54 }
55}
56
57#[derive(Clone, Debug, Eq, PartialEq)]
63pub struct RawLineRecord {
64 pub content: Vec<u8>,
65 pub terminator: TerminatorKind,
66}
67
68pub type RawLine = RawLineRecord;
69
70impl RawLineRecord {
71 pub fn new(content: Vec<u8>, terminator: TerminatorKind) -> Self {
72 Self {
73 content,
74 terminator,
75 }
76 }
77
78 pub fn content(&self) -> &[u8] {
79 &self.content
80 }
81
82 pub fn terminator(&self) -> TerminatorKind {
83 self.terminator
84 }
85
86 pub fn to_bytes(&self) -> Vec<u8> {
88 let mut bytes = self.content.clone();
89 match self.terminator {
90 TerminatorKind::Lf => bytes.push(b'\n'),
91 TerminatorKind::CrLf => bytes.extend_from_slice(b"\r\n"),
92 TerminatorKind::None => {}
93 }
94 bytes
95 }
96}
97
98#[derive(Clone, Debug, Eq, PartialEq)]
100pub struct RetainedLine {
101 pub line_number: usize,
102 pub record: RawLineRecord,
103 pub byte_start: u64,
105 pub byte_end: u64,
108}
109
110impl RetainedLine {
111 pub fn content(&self) -> &[u8] {
112 self.record.content()
113 }
114
115 pub fn terminator(&self) -> TerminatorKind {
116 self.record.terminator()
117 }
118}
119
120#[derive(Clone, Debug, Eq, PartialEq)]
122pub struct CoverageInput {
123 pub requested_lines: BTreeSet<usize>,
126 pub retain_all: bool,
128}
129
130impl Default for CoverageInput {
131 fn default() -> Self {
132 Self::none()
133 }
134}
135
136impl CoverageInput {
137 pub fn none() -> Self {
138 Self {
139 requested_lines: BTreeSet::new(),
140 retain_all: false,
141 }
142 }
143
144 pub fn whole_file() -> Self {
145 Self {
146 requested_lines: BTreeSet::new(),
147 retain_all: true,
148 }
149 }
150
151 pub fn line(line_number: usize) -> Self {
152 Self::lines([line_number])
153 }
154
155 pub fn lines<I>(lines: I) -> Self
156 where
157 I: IntoIterator<Item = usize>,
158 {
159 Self {
160 requested_lines: lines.into_iter().collect(),
161 retain_all: false,
162 }
163 }
164
165 pub fn range(first: usize, last: usize) -> Self {
166 if first > last {
167 return Self::none();
168 }
169 Self {
170 requested_lines: (first..=last).collect(),
171 retain_all: false,
172 }
173 }
174
175 pub fn retains(&self, line_number: usize) -> bool {
176 self.retain_all || self.requested_lines.contains(&line_number)
177 }
178
179 pub fn is_whole_file(&self) -> bool {
180 self.retain_all
181 }
182}
183
184#[derive(Clone, Debug, Default, Eq, PartialEq)]
187pub struct CaptureProvenance {
188 pub source: Option<PathBuf>,
189 pub file_identity: Option<FileIdentity>,
190 pub byte_len: Option<u64>,
191 pub modified: Option<SystemTime>,
192 pub labels: BTreeMap<String, String>,
195}
196
197impl CaptureProvenance {
198 pub fn from_source(source: impl Into<PathBuf>) -> Self {
199 Self {
200 source: Some(source.into()),
201 ..Self::default()
202 }
203 }
204
205 pub fn with_source(mut self, source: impl Into<PathBuf>) -> Self {
206 self.source = Some(source.into());
207 self
208 }
209
210 pub fn with_byte_len(mut self, byte_len: u64) -> Self {
211 self.byte_len = Some(byte_len);
212 self
213 }
214
215 pub fn with_file_identity(mut self, file_identity: FileIdentity) -> Self {
216 self.file_identity = Some(file_identity);
217 self
218 }
219
220 pub fn with_modified(mut self, modified: SystemTime) -> Self {
221 self.modified = Some(modified);
222 self
223 }
224
225 pub fn with_label(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
226 self.labels.insert(key.into(), value.into());
227 self
228 }
229
230 pub fn from_metadata(source: impl Into<PathBuf>, metadata: &Metadata) -> Self {
234 Self {
235 source: Some(source.into()),
236 file_identity: file_identity(metadata),
237 byte_len: Some(metadata.len()),
238 modified: metadata.modified().ok(),
239 labels: BTreeMap::new(),
240 }
241 }
242
243 pub fn same_file_version(&self, other: &Self) -> bool {
246 if self.file_identity.is_some()
247 && other.file_identity.is_some()
248 && self.file_identity != other.file_identity
249 {
250 return false;
251 }
252 if self.byte_len.is_some() && other.byte_len.is_some() && self.byte_len != other.byte_len {
253 return false;
254 }
255 if self.modified.is_some() && other.modified.is_some() && self.modified != other.modified {
256 return false;
257 }
258 true
259 }
260}
261
262#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
265pub struct FileIdentity {
266 pub device: u64,
267 pub inode: u64,
268}
269
270#[cfg(unix)]
271fn file_identity(metadata: &Metadata) -> Option<FileIdentity> {
272 use std::os::unix::fs::MetadataExt;
273
274 Some(FileIdentity {
275 device: metadata.dev(),
276 inode: metadata.ino(),
277 })
278}
279
280#[cfg(not(unix))]
281fn file_identity(_metadata: &Metadata) -> Option<FileIdentity> {
282 None
283}
284
285#[derive(Clone, Debug, Eq, PartialEq)]
287pub struct ScanRequest {
288 pub coverage: CoverageInput,
289 pub provenance: CaptureProvenance,
290}
291
292impl Default for ScanRequest {
293 fn default() -> Self {
294 Self::whole_file()
295 }
296}
297
298impl ScanRequest {
299 pub fn new(coverage: CoverageInput) -> Self {
300 Self {
301 coverage,
302 provenance: CaptureProvenance::default(),
303 }
304 }
305
306 pub fn whole_file() -> Self {
307 Self::new(CoverageInput::whole_file())
308 }
309
310 pub fn for_lines<I>(lines: I) -> Self
311 where
312 I: IntoIterator<Item = usize>,
313 {
314 Self::new(CoverageInput::lines(lines))
315 }
316
317 pub fn for_range(first: usize, last: usize) -> Self {
318 Self::new(CoverageInput::range(first, last))
319 }
320
321 pub fn with_provenance(mut self, provenance: CaptureProvenance) -> Self {
322 self.provenance = provenance;
323 self
324 }
325}
326
327#[derive(Clone, Debug, Eq, PartialEq)]
330pub struct ScanCoverage {
331 pub requested_lines: BTreeSet<usize>,
332 pub retain_all: bool,
333 pub retained_lines: BTreeSet<usize>,
334 pub seen_lines: BTreeSet<usize>,
335 pub scanned_line_count: usize,
336 pub total_lines: usize,
337 pub byte_count: u64,
338 pub eof_observed: bool,
339}
340
341pub type Coverage = ScanCoverage;
342
343impl ScanCoverage {
344 pub fn is_seen(&self, line_number: usize) -> bool {
345 self.seen_lines.contains(&line_number)
346 }
347
348 pub fn is_retained(&self, line_number: usize) -> bool {
349 self.retained_lines.contains(&line_number)
350 }
351
352 pub fn is_eligible(&self, line_number: usize) -> bool {
353 self.is_seen(line_number)
354 }
355
356 pub fn scanned_to_eof(&self) -> bool {
357 self.eof_observed
358 }
359}
360
361#[derive(Clone, Debug, Eq, PartialEq)]
364pub struct BoundaryEvidence {
365 pub empty_file: bool,
366 pub bof_observed: bool,
367 pub eof_observed: bool,
368 pub first_seen: Option<usize>,
369 pub last_seen: Option<usize>,
370}
371
372#[derive(Clone, Debug, Eq, PartialEq)]
374pub struct Snapshot {
375 pub tag: String,
376 pub normalized_bytes: Vec<u8>,
377 pub records: BTreeMap<usize, RawLineRecord>,
378 pub retained_lines: BTreeMap<usize, RetainedLine>,
379 pub coverage: ScanCoverage,
380 pub boundary: BoundaryEvidence,
381 pub total_lines: usize,
382 pub byte_count: u64,
383 pub provenance: CaptureProvenance,
385 pub capture_provenance: CaptureProvenance,
387}
388
389impl Snapshot {
390 pub fn raw_record(&self, line_number: usize) -> Option<&RawLineRecord> {
391 self.records.get(&line_number)
392 }
393
394 pub fn retained_line(&self, line_number: usize) -> Option<&RetainedLine> {
395 self.retained_lines.get(&line_number)
396 }
397
398 pub fn is_seen(&self, line_number: usize) -> bool {
399 self.coverage.is_seen(line_number)
400 }
401
402 pub fn eof_observed(&self) -> bool {
403 self.coverage.eof_observed
404 }
405
406 pub fn verify_record(&self, line_number: usize, expected: &RawLineRecord) -> bool {
407 self.raw_record(line_number) == Some(expected)
408 }
409}
410
411#[derive(Clone, Debug, Eq, PartialEq)]
414pub struct ScanResult {
415 pub snapshot: Option<Snapshot>,
416 pub coverage: ScanCoverage,
417 pub provenance: CaptureProvenance,
418 pub capture_provenance: CaptureProvenance,
419}
420
421impl ScanResult {
422 pub fn published_snapshot(&self) -> Option<&Snapshot> {
423 self.snapshot.as_ref()
424 }
425
426 pub fn into_snapshot(self) -> Option<Snapshot> {
427 self.snapshot
428 }
429}
430
431#[derive(Debug)]
433pub enum ScanError {
434 Io(io::Error),
435 AlreadyFinished,
436}
437
438impl fmt::Display for ScanError {
439 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
440 match self {
441 Self::Io(error) => write!(formatter, "hashline scan I/O error: {error}"),
442 Self::AlreadyFinished => formatter.write_str("hashline scan already reached EOF"),
443 }
444 }
445}
446
447impl std::error::Error for ScanError {
448 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
449 match self {
450 Self::Io(error) => Some(error),
451 Self::AlreadyFinished => None,
452 }
453 }
454}
455
456impl From<io::Error> for ScanError {
457 fn from(error: io::Error) -> Self {
458 Self::Io(error)
459 }
460}
461
462pub struct ForwardScanner {
468 request: ScanRequest,
469 capture_provenance: CaptureProvenance,
470 raw_bytes: Vec<u8>,
471 current_line: Vec<u8>,
472 current_line_start: u64,
473 next_line_number: usize,
474 scanned_line_count: usize,
475 retained_lines: BTreeMap<usize, RetainedLine>,
476 finished: bool,
477 published: Option<Snapshot>,
478}
479
480impl ForwardScanner {
481 pub fn new(request: ScanRequest) -> Self {
482 let capture_provenance = CaptureProvenance::default();
483 Self {
484 request,
485 capture_provenance,
486 raw_bytes: Vec::new(),
487 current_line: Vec::new(),
488 current_line_start: 0,
489 next_line_number: 1,
490 scanned_line_count: 0,
491 retained_lines: BTreeMap::new(),
492 finished: false,
493 published: None,
494 }
495 }
496
497 pub fn with_capture_provenance(
498 request: ScanRequest,
499 capture_provenance: CaptureProvenance,
500 ) -> Self {
501 let mut scanner = Self::new(request);
502 scanner.capture_provenance = capture_provenance;
503 scanner
504 }
505
506 pub fn push(&mut self, bytes: &[u8]) -> Result<(), ScanError> {
509 if self.finished {
510 return Err(ScanError::AlreadyFinished);
511 }
512
513 for &byte in bytes {
514 self.raw_bytes.push(byte);
515 if byte == b'\n' {
516 let mut content = std::mem::take(&mut self.current_line);
517 let terminator = if content.last() == Some(&b'\r') {
518 content.pop();
519 TerminatorKind::CrLf
520 } else {
521 TerminatorKind::Lf
522 };
523 self.scanned_line_count += 1;
524 let line_number = self.next_line_number;
525 self.next_line_number += 1;
526 let byte_end = self.raw_bytes.len() as u64;
527 self.retain_line_if_requested(
528 line_number,
529 RawLineRecord::new(content, terminator),
530 self.current_line_start,
531 byte_end,
532 );
533 self.current_line_start = byte_end;
534 } else {
535 self.current_line.push(byte);
536 }
537 }
538 Ok(())
539 }
540
541 pub fn published_snapshot(&self) -> Option<&Snapshot> {
543 self.published.as_ref()
544 }
545
546 pub fn eof_observed(&self) -> bool {
547 self.finished
548 }
549
550 pub fn finish(&mut self) -> Result<ScanResult, ScanError> {
552 if self.finished {
553 return Err(ScanError::AlreadyFinished);
554 }
555
556 if !self.current_line.is_empty() {
557 self.scanned_line_count += 1;
558 let line_number = self.next_line_number;
559 let byte_end = self.raw_bytes.len() as u64;
560 let content = std::mem::take(&mut self.current_line);
561 self.retain_line_if_requested(
562 line_number,
563 RawLineRecord::new(content, TerminatorKind::None),
564 self.current_line_start,
565 byte_end,
566 );
567 }
568
569 self.finished = true;
570 let requested_lines = self.request.coverage.requested_lines.clone();
571 let retained_line_numbers: BTreeSet<usize> = self.retained_lines.keys().copied().collect();
572 let coverage = ScanCoverage {
573 requested_lines,
574 retain_all: self.request.coverage.retain_all,
575 retained_lines: retained_line_numbers.clone(),
576 seen_lines: retained_line_numbers,
577 scanned_line_count: self.scanned_line_count,
578 total_lines: self.scanned_line_count,
579 byte_count: self.raw_bytes.len() as u64,
580 eof_observed: true,
581 };
582 let first_seen = self.retained_lines.keys().next().copied();
583 let last_seen = self.retained_lines.keys().next_back().copied();
584 let boundary = BoundaryEvidence {
585 empty_file: self.raw_bytes.is_empty(),
586 bof_observed: true,
587 eof_observed: true,
588 first_seen,
589 last_seen,
590 };
591 let records = self
592 .retained_lines
593 .iter()
594 .map(|(&line_number, retained)| (line_number, retained.record.clone()))
595 .collect();
596 let normalized_bytes = normalize_for_tag(&self.raw_bytes);
597 let mut capture_provenance = self.capture_provenance.clone();
598 capture_provenance.byte_len = Some(self.raw_bytes.len() as u64);
599 let snapshot = Snapshot {
600 tag: tag_for(&self.raw_bytes),
601 normalized_bytes,
602 records,
603 retained_lines: self.retained_lines.clone(),
604 coverage: coverage.clone(),
605 boundary,
606 total_lines: self.scanned_line_count,
607 byte_count: self.raw_bytes.len() as u64,
608 provenance: self.request.provenance.clone(),
609 capture_provenance: capture_provenance.clone(),
610 };
611 self.published = Some(snapshot.clone());
612 Ok(ScanResult {
613 snapshot: Some(snapshot),
614 coverage,
615 provenance: self.request.provenance.clone(),
616 capture_provenance,
617 })
618 }
619
620 fn retain_line_if_requested(
621 &mut self,
622 line_number: usize,
623 record: RawLineRecord,
624 byte_start: u64,
625 byte_end: u64,
626 ) {
627 if self.request.coverage.retains(line_number) {
628 self.retained_lines.insert(
629 line_number,
630 RetainedLine {
631 line_number,
632 record,
633 byte_start,
634 byte_end,
635 },
636 );
637 }
638 }
639}
640
641pub fn scan_reader<R: Read>(reader: &mut R, request: ScanRequest) -> Result<ScanResult, ScanError> {
643 scan_reader_with_provenance(reader, request, CaptureProvenance::default())
644}
645
646pub fn scan_reader_with_provenance<R: Read>(
650 reader: &mut R,
651 request: ScanRequest,
652 capture_provenance: CaptureProvenance,
653) -> Result<ScanResult, ScanError> {
654 let mut scanner = ForwardScanner::with_capture_provenance(request, capture_provenance);
655 let mut buffer = [0_u8; 16 * 1024];
656 loop {
657 let read = reader.read(&mut buffer)?;
658 if read == 0 {
659 break;
660 }
661 scanner.push(&buffer[..read])?;
662 }
663 scanner.finish()
664}
665
666pub fn scan_bytes(bytes: &[u8]) -> Snapshot {
668 scan_bytes_with_request(bytes, ScanRequest::whole_file())
669 .snapshot
670 .expect("a byte slice scan always observes EOF")
671}
672
673pub fn scan_bytes_with_request(bytes: &[u8], request: ScanRequest) -> ScanResult {
675 let capture_provenance = request.provenance.clone().with_byte_len(bytes.len() as u64);
676 let mut scanner = ForwardScanner::with_capture_provenance(request, capture_provenance);
677 scanner
678 .push(bytes)
679 .expect("a new scanner cannot be finished");
680 scanner
681 .finish()
682 .expect("a byte slice scan can finish exactly once")
683}
684
685pub fn scan_bytes_with_coverage(bytes: &[u8], coverage: CoverageInput) -> Snapshot {
686 scan_bytes_with_request(bytes, ScanRequest::new(coverage))
687 .snapshot
688 .expect("a byte slice scan always observes EOF")
689}
690
691pub fn scan(bytes: &[u8], request: ScanRequest) -> ScanResult {
694 scan_bytes_with_request(bytes, request)
695}
696
697pub fn raw_line_records(bytes: &[u8]) -> Vec<RawLineRecord> {
699 let snapshot = scan_bytes(bytes);
700 snapshot.records.into_values().collect()
701}
702
703pub fn normalize_whole_file(bytes: &[u8]) -> Vec<u8> {
705 normalize_for_tag(bytes)
706}
707
708pub fn whole_file_tag(bytes: &[u8]) -> String {
710 tag_for(bytes)
711}
712
713pub fn capture_path(
718 path: impl AsRef<Path>,
719 request: ScanRequest,
720) -> Result<CoherentCapture, CaptureError> {
721 let path = path.as_ref().to_path_buf();
722 let mut last_unstable = None;
723 for attempt in 1..=2 {
724 match capture_path_once(&path, &request) {
725 Ok(snapshot) => {
726 return Ok(CoherentCapture {
727 snapshot,
728 attempts: attempt,
729 });
730 }
731 Err(CaptureError::Unstable {
732 before,
733 after,
734 observed_bytes,
735 }) if attempt == 1 => {
736 last_unstable = Some((before, after, observed_bytes));
737 }
738 Err(error) => return Err(error),
739 }
740 }
741 let (before, after, observed_bytes) =
742 last_unstable.expect("the retry loop records instability");
743 Err(CaptureError::Unstable {
744 before,
745 after,
746 observed_bytes,
747 })
748}
749
750pub fn scan_path(path: impl AsRef<Path>, request: ScanRequest) -> Result<Snapshot, CaptureError> {
752 capture_path(path, request).map(|capture| capture.snapshot)
753}
754
755#[derive(Clone, Debug, Eq, PartialEq)]
756pub struct CoherentCapture {
757 pub snapshot: Snapshot,
758 pub attempts: u8,
759}
760
761#[derive(Debug)]
762pub enum CaptureError {
763 Io(io::Error),
764 Unstable {
765 before: CaptureProvenance,
766 after: CaptureProvenance,
767 observed_bytes: u64,
768 },
769}
770
771impl fmt::Display for CaptureError {
772 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
773 match self {
774 Self::Io(error) => write!(formatter, "hashline capture I/O error: {error}"),
775 Self::Unstable {
776 before,
777 after,
778 observed_bytes,
779 } => write!(
780 formatter,
781 "hashline capture was unstable (before={before:?}, after={after:?}, observed_bytes={observed_bytes})"
782 ),
783 }
784 }
785}
786
787impl std::error::Error for CaptureError {
788 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
789 match self {
790 Self::Io(error) => Some(error),
791 Self::Unstable { .. } => None,
792 }
793 }
794}
795
796impl From<io::Error> for CaptureError {
797 fn from(error: io::Error) -> Self {
798 Self::Io(error)
799 }
800}
801
802impl From<ScanError> for CaptureError {
803 fn from(error: ScanError) -> Self {
804 match error {
805 ScanError::Io(error) => Self::Io(error),
806 ScanError::AlreadyFinished => Self::Io(io::Error::new(
807 io::ErrorKind::InvalidData,
808 "hashline scan finished before capture completed",
809 )),
810 }
811 }
812}
813
814fn capture_path_once(path: &Path, request: &ScanRequest) -> Result<Snapshot, CaptureError> {
815 let mut file = File::open(path)?;
816 let before = CaptureProvenance::from_metadata(path.to_path_buf(), &file.metadata()?);
817 let result = scan_reader_with_provenance(&mut file, request.clone(), before.clone())?;
818 let after_descriptor = CaptureProvenance::from_metadata(path.to_path_buf(), &file.metadata()?);
819 let after_path =
824 CaptureProvenance::from_metadata(path.to_path_buf(), &std::fs::metadata(path)?);
825 let observed_bytes = result.coverage.byte_count;
826 let stable = before.same_file_version(&after_descriptor)
827 && before.same_file_version(&after_path)
828 && before.byte_len == Some(observed_bytes)
829 && after_descriptor.byte_len == Some(observed_bytes)
830 && after_path.byte_len == Some(observed_bytes);
831 if !stable {
832 return Err(CaptureError::Unstable {
833 before,
834 after: after_path,
835 observed_bytes,
836 });
837 }
838 result.into_snapshot().ok_or(CaptureError::Unstable {
839 before,
840 after: after_path,
841 observed_bytes,
842 })
843}
844
845#[cfg(test)]
846mod tests {
847 use super::*;
848 use base64::Engine as _;
849 use serde_json::Value;
850 use std::io::Cursor;
851 use std::io::Write;
852 use tempfile::NamedTempFile;
853
854 #[test]
855 fn raw_records_preserve_bom_terminators_and_interior_carriage_returns() {
856 let bytes = b"\xEF\xBB\xBFfirst\r\nsecond\rinterior\nlast";
857 let records = raw_line_records(bytes);
858 assert_eq!(records.len(), 3);
859 assert_eq!(records[0].content, b"\xEF\xBB\xBFfirst");
860 assert_eq!(records[0].terminator, TerminatorKind::CrLf);
861 assert_eq!(records[1].content, b"second\rinterior");
862 assert_eq!(records[1].terminator, TerminatorKind::Lf);
863 assert_eq!(records[2].content, b"last");
864 assert_eq!(records[2].terminator, TerminatorKind::None);
865 }
866
867 #[test]
868 fn a_trailing_terminator_does_not_create_an_extra_record() {
869 assert_eq!(raw_line_records(b"one\n").len(), 1);
870 assert_eq!(raw_line_records(b"one\r\n").len(), 1);
871 assert!(raw_line_records(b"").is_empty());
872 }
873
874 #[test]
875 fn normalized_tag_ignores_only_trailing_space_tab_and_cr() {
876 let lf = b"alpha\nbeta\n";
877 let decorated = b"alpha \t\r\nbeta\r\n";
878 assert_eq!(normalize_whole_file(decorated), normalize_whole_file(lf));
879 assert_eq!(whole_file_tag(decorated), whole_file_tag(lf));
880 assert_eq!(normalize_whole_file(b"a\rreturn\n"), b"a\rreturn\n");
881 assert_eq!(normalize_whole_file(b"a \t\r"), b"a");
882 }
883
884 #[test]
885 fn ranged_retention_uses_the_same_raw_parser_as_whole_file_retention() {
886 let bytes = b"zero\r\none\ntwo\rthree\n";
887 let whole = scan_bytes(bytes);
888 let ranged = scan_bytes_with_request(bytes, ScanRequest::for_range(2, 2));
889 let ranged_snapshot = ranged.snapshot.unwrap();
890 assert_eq!(ranged_snapshot.raw_record(2), whole.raw_record(2));
891 assert_eq!(ranged_snapshot.coverage.scanned_line_count, 3);
892 assert!(!ranged_snapshot.is_seen(1));
893 assert!(!ranged_snapshot.is_seen(3));
894 assert_eq!(ranged_snapshot.total_lines, whole.total_lines);
895 assert_eq!(ranged_snapshot.tag, whole.tag);
896 }
897
898 #[test]
899 fn scan_only_rows_are_not_seen_or_eligible() {
900 let result = scan_bytes_with_request(b"a\nb\nc\n", ScanRequest::for_lines([2]));
901 let snapshot = result.snapshot.unwrap();
902 assert_eq!(snapshot.coverage.scanned_line_count, 3);
903 assert_eq!(snapshot.coverage.seen_lines, BTreeSet::from([2]));
904 assert!(snapshot.coverage.is_eligible(2));
905 assert!(!snapshot.coverage.is_eligible(3));
906 assert!(snapshot.raw_record(3).is_none());
907 }
908
909 #[test]
910 fn no_snapshot_is_published_before_eof() {
911 let mut scanner = ForwardScanner::new(ScanRequest::for_lines([1]));
912 scanner.push(b"first\nsecond").unwrap();
913 assert!(!scanner.eof_observed());
914 assert!(scanner.published_snapshot().is_none());
915 let result = scanner.finish().unwrap();
916 assert!(result.coverage.eof_observed);
917 assert!(scanner.published_snapshot().is_some());
918 assert!(scanner.push(b"after eof").is_err());
919 }
920
921 #[test]
922 fn reader_scan_is_forward_and_publishes_at_eof() {
923 let mut reader = Cursor::new(b"one\r\ntwo\nthree".to_vec());
924 let result = scan_reader(&mut reader, ScanRequest::for_lines([2])).unwrap();
925 let snapshot = result.snapshot.unwrap();
926 assert_eq!(
927 snapshot.raw_record(2).unwrap().terminator,
928 TerminatorKind::Lf
929 );
930 assert_eq!(snapshot.total_lines, 3);
931 assert!(snapshot.eof_observed());
932 }
933
934 #[test]
935 fn path_capture_exposes_provenance_and_observes_eof() {
936 let mut file = NamedTempFile::new().unwrap();
937 file.write_all(b"one\ntwo").unwrap();
938 file.flush().unwrap();
939 let request = ScanRequest::for_lines([2])
940 .with_provenance(CaptureProvenance::default().with_label("reader", "test"));
941 let capture = capture_path(file.path(), request).unwrap();
942 assert_eq!(capture.attempts, 1);
943 assert_eq!(capture.snapshot.total_lines, 2);
944 assert_eq!(capture.snapshot.provenance.labels["reader"], "test");
945 assert_eq!(capture.snapshot.capture_provenance.byte_len, Some(7));
946 }
947
948 #[test]
949 fn oracle_corpus_byte_model_rows() {
950 const OWNED: &[&str] = &[
957 "lf",
958 "lf-rejection",
959 "crlf",
960 "crlf-rejection",
961 "mixed-terminators",
962 "mixed-terminators-rejection",
963 "bom",
964 "bom-rejection",
965 "empty",
966 "empty-rejection",
967 "missing-final-newline",
968 "missing-final-newline-rejection",
969 "unicode",
970 "unicode-rejection",
971 "trailing-whitespace",
972 "trailing-whitespace-rejection",
973 ];
974 const DEFERRED: &[&str] = &[
975 "bof",
976 "bof-rejection",
977 "eof",
978 "eof-rejection",
979 "eof-relative",
980 "eof-relative-rejection",
981 "one-line",
982 "one-line-rejection",
983 "empty-boundary",
984 "empty-boundary-rejection",
985 "block",
986 "block-rejection",
987 "repair",
988 "repair-negative-control",
989 "registered-deviation",
990 "registered-deviation-negative-control",
991 "named-register",
992 "anonymous-register",
993 "cross-file-register",
994 "register-overflow",
995 ];
996
997 let mut consumed = 0;
998 let mut deferred = 0;
999 for line in include_str!("../oracle/fixtures.jsonl").lines() {
1000 let row: Value = serde_json::from_str(line).expect("oracle fixture JSON must parse");
1001 let category = row["fixture_category"]
1002 .as_str()
1003 .expect("oracle fixture category must be a string");
1004 if !OWNED.contains(&category) {
1005 assert!(
1006 DEFERRED.contains(&category),
1007 "new oracle category {category:?} needs an explicit slice owner"
1008 );
1009 deferred += 1;
1010 continue;
1011 }
1012 consumed += 1;
1013
1014 let bytes = base64::engine::general_purpose::STANDARD
1015 .decode(row["initial_base64"].as_str().unwrap())
1016 .expect("oracle fixture initial_base64 must decode");
1017 let snapshot = scan_bytes(&bytes);
1018 let expected_tag = row["snapshot_tag"].as_str().unwrap();
1019 assert_eq!(snapshot.tag, expected_tag, "fixture {}", row["id"]);
1020 assert_eq!(snapshot.tag, tag_for(&bytes), "fixture {}", row["id"]);
1021
1022 let family = category.strip_suffix("-rejection").unwrap_or(category);
1023 let expected_terminators: Vec<TerminatorKind> = match family {
1024 "lf" | "unicode" => vec![TerminatorKind::Lf; 3],
1025 "bom" => vec![TerminatorKind::Lf; 2],
1026 "crlf" => vec![TerminatorKind::CrLf; 3],
1027 "mixed-terminators" => vec![
1028 TerminatorKind::Lf,
1029 TerminatorKind::CrLf,
1030 TerminatorKind::None,
1031 ],
1032 "empty" => Vec::new(),
1033 "missing-final-newline" => vec![TerminatorKind::Lf, TerminatorKind::None],
1034 "trailing-whitespace" => vec![TerminatorKind::Lf, TerminatorKind::Lf],
1035 _ => unreachable!("owned category must have a byte-model shape"),
1036 };
1037 let records: Vec<&RawLineRecord> = snapshot.records.values().collect();
1038 assert_eq!(
1039 records.len(),
1040 expected_terminators.len(),
1041 "fixture {}",
1042 row["id"]
1043 );
1044 assert_eq!(
1045 snapshot.total_lines,
1046 expected_terminators.len(),
1047 "fixture {}",
1048 row["id"]
1049 );
1050 assert_eq!(
1051 records
1052 .iter()
1053 .map(|record| record.terminator)
1054 .collect::<Vec<_>>(),
1055 expected_terminators,
1056 "fixture {}",
1057 row["id"]
1058 );
1059 let reconstructed: Vec<u8> = records
1060 .iter()
1061 .flat_map(|record| record.to_bytes())
1062 .collect();
1063 assert_eq!(reconstructed, bytes, "fixture {}", row["id"]);
1064 }
1065
1066 assert_eq!(
1067 consumed, 64,
1068 "the owned byte-model corpus must remain complete"
1069 );
1070 assert_eq!(
1071 deferred, 64,
1072 "the deferred corpus rows must remain explicitly classified"
1073 );
1074 }
1075}