spate-s3 0.1.0

Coordinated object-storage (S3) backfill source for the Spate pipeline framework: an elected leader plans the prefix into splits, workers lease them with fenced per-split progress in a coordination store, resume is drift-checked against ETag pins, and the job self-terminates when the plan completes. Applications should depend on the `spate` facade crate with the `s3` feature.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

//! Test-only framer shared across the crate's unit tests.
//!
//! `spate-s3` ships no record framer of its own — production wires the format's
//! framer via [`S3Source::with_framer`](crate::S3Source::with_framer). The
//! tests still need to turn newline-delimited object bytes into records, so
//! they carry this local line framer, mirroring `spate-json`'s `NdjsonFramer`
//! semantics without taking a dependency on a format crate.

use spate_core::framing::RecordFramer;
use std::collections::VecDeque;
use std::io;

/// Newline-delimited [`RecordFramer`]: split on `\n`, strip one trailing `\r`,
/// skip whitespace-only lines, keep an unterminated final line, bounded per
/// record. Behaviourally identical to `spate-json`'s `NdjsonFramer`, so the
/// resume-by-record-index tests exercise the same record boundaries.
pub(crate) struct TestLineFramer {
    partial: Vec<u8>,
    ready: VecDeque<Vec<u8>>,
    decoded_bytes: u64,
    max_record_bytes: usize,
}

impl TestLineFramer {
    pub(crate) fn new(max_record_bytes: usize) -> TestLineFramer {
        TestLineFramer {
            partial: Vec::new(),
            ready: VecDeque::new(),
            decoded_bytes: 0,
            max_record_bytes,
        }
    }

    fn push_partial(&mut self, bytes: &[u8]) -> io::Result<()> {
        if self.partial.len() + bytes.len() > self.max_record_bytes {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!(
                    "record line exceeds the configured max_record_bytes ({})",
                    self.max_record_bytes
                ),
            ));
        }
        self.partial.extend_from_slice(bytes);
        Ok(())
    }

    fn complete_line(&mut self) {
        if self.partial.last() == Some(&b'\r') {
            self.partial.pop();
        }
        if self.partial.iter().all(u8::is_ascii_whitespace) {
            self.partial.clear();
            return;
        }
        self.ready.push_back(std::mem::take(&mut self.partial));
    }
}

impl RecordFramer for TestLineFramer {
    fn push(&mut self, bytes: &[u8]) -> io::Result<()> {
        self.decoded_bytes += bytes.len() as u64;
        let mut rest = bytes;
        while let Some(nl) = rest.iter().position(|&b| b == b'\n') {
            self.push_partial(&rest[..nl])?;
            self.complete_line();
            rest = &rest[nl + 1..];
        }
        self.push_partial(rest)
    }

    fn finish(&mut self) -> io::Result<()> {
        if !self.partial.is_empty() {
            self.complete_line();
        }
        Ok(())
    }

    fn pop(&mut self) -> Option<Vec<u8>> {
        self.ready.pop_front()
    }

    fn decoded_bytes(&self) -> u64 {
        self.decoded_bytes
    }
}