rupg_types/
snapshot.rs

1//! `pg_snapshot` and `txid_snapshot`, in the text and binary formats of `src/backend/utils/adt/xid8funcs.c`. A value is the binary form: the number of transactions in progress as an `int4`, then `xmin`, `xmax` and each transaction in progress as an `int8`.
2
3use rupg_common::SqlState;
4
5use crate::binary::Recv;
6use crate::error::TypeError;
7use crate::number::is_space;
8
9/// `PG_SNAPSHOT_MAX_NXIP`: the most transactions in progress that fit in a value of `MaxAllocSize` bytes.
10const MAX_NXIP: usize = (0x3fff_ffff - 20) / 8;
11
12/// A snapshot: the first transaction that is still in progress, the first transaction that did not start, and the transactions in progress between them, in order and with no duplicates.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct Snapshot {
15    pub xmin: u64,
16    pub xmax: u64,
17    pub xip: Vec<u64>,
18}
19
20impl Snapshot {
21    /// The binary form, which is also the value.
22    #[must_use]
23    pub fn to_bytes(&self) -> Vec<u8> {
24        let mut out = Vec::with_capacity(20 + 8 * self.xip.len());
25        let nxip = i32::try_from(self.xip.len()).unwrap_or(i32::MAX);
26        out.extend_from_slice(&nxip.to_be_bytes());
27        out.extend_from_slice(&self.xmin.to_be_bytes());
28        out.extend_from_slice(&self.xmax.to_be_bytes());
29        for xid in &self.xip {
30            out.extend_from_slice(&xid.to_be_bytes());
31        }
32        out
33    }
34
35    /// The snapshot in the bytes that [`Snapshot::to_bytes`] gives, or `None` for bytes of another form.
36    #[must_use]
37    pub fn from_bytes(bytes: &[u8]) -> Option<Snapshot> {
38        let mut recv = Recv::new(bytes);
39        let nxip = usize::try_from(recv.i32().ok()?).ok()?;
40        let xmin = recv.i64().ok()?.cast_unsigned();
41        let xmax = recv.i64().ok()?.cast_unsigned();
42        if recv.remaining() != nxip * 8 {
43            return None;
44        }
45        let xip =
46            (0..nxip).map(|_| recv.i64().map(i64::cast_unsigned)).collect::<Result<_, _>>().ok()?;
47        Some(Snapshot { xmin, xmax, xip })
48    }
49
50    /// `is_visible_fxid`: true when the transaction ended before the snapshot. A transaction before `xmin` ended, a transaction at `xmax` or after it did not start, and a transaction between them ended when it is not in progress.
51    #[must_use]
52    pub fn is_visible(&self, xid: u64) -> bool {
53        if xid < self.xmin {
54            return true;
55        }
56        xid < self.xmax && self.xip.binary_search(&xid).is_err()
57    }
58}
59
60/// `strtou64(s, &end, 10)`: the value and the length of the number at the start of `b`, which is 0 when there is no number. `strtoull` skips the spaces before the number, gives `u64::MAX` for a number that overflows and negates a number after a `-`.
61fn strtou64(b: &[u8]) -> (u64, usize) {
62    let mut i = 0;
63    while b.get(i).is_some_and(|&c| is_space(c)) {
64        i += 1;
65    }
66    let neg = b.get(i) == Some(&b'-');
67    if neg || b.get(i) == Some(&b'+') {
68        i += 1;
69    }
70    let first = i;
71    let mut value = Some(0u64);
72    while let Some(&c) = b.get(i).filter(|c| c.is_ascii_digit()) {
73        value =
74            value.and_then(|v| v.checked_mul(10)).and_then(|v| v.checked_add(u64::from(c - b'0')));
75        i += 1;
76    }
77    if i == first {
78        return (0, 0);
79    }
80    match value {
81        None => (u64::MAX, i),
82        Some(v) if neg => (v.wrapping_neg(), i),
83        Some(v) => (v, i),
84    }
85}
86
87/// `pg_snapshot_in`: `xmin:xmax:` and the transactions in progress with a `,` between them. `xmin` is not 0, `xmax` is not before `xmin`, and each transaction in progress is in order from `xmin` up to `xmax`. A duplicate is skipped. The error names `pg_snapshot` also for `txid_snapshot`.
88///
89/// # Errors
90///
91/// `22P02` for text that does not have this form.
92pub fn snapshot_in(s: &str) -> Result<Snapshot, TypeError> {
93    let syntax = || TypeError::syntax("pg_snapshot", s);
94    let b = s.as_bytes();
95    let (xmin, end) = strtou64(b);
96    if b.get(end) != Some(&b':') {
97        return Err(syntax());
98    }
99    let mut at = end + 1;
100    let (xmax, end) = strtou64(&b[at..]);
101    if b.get(at + end) != Some(&b':') {
102        return Err(syntax());
103    }
104    at += end + 1;
105    if xmin == 0 || xmax == 0 || xmax < xmin {
106        return Err(syntax());
107    }
108    let mut xip = Vec::new();
109    let mut last = 0;
110    while at < b.len() {
111        let (xid, end) = strtou64(&b[at..]);
112        at += end;
113        if xid < xmin || xid >= xmax || xid < last {
114            return Err(syntax());
115        }
116        if xid != last {
117            xip.push(xid);
118        }
119        last = xid;
120        match b.get(at) {
121            Some(b',') => at += 1,
122            Some(_) => return Err(syntax()),
123            None => {}
124        }
125    }
126    Ok(Snapshot { xmin, xmax, xip })
127}
128
129/// `pg_snapshot_out`: `xmin:xmax:` and the transactions in progress with a `,` between them.
130pub fn snapshot_out(snapshot: &Snapshot, out: &mut Vec<u8>) {
131    let mut text = format!("{}:{}:", snapshot.xmin, snapshot.xmax);
132    for (i, xid) in snapshot.xip.iter().enumerate() {
133        if i > 0 {
134            text.push(',');
135        }
136        text.push_str(&xid.to_string());
137    }
138    out.extend_from_slice(text.as_bytes());
139}
140
141/// `pg_snapshot_recv`: the binary form, with the checks of the text input. A transaction in progress can be equal to `xmax` here, as in PostgreSQL.
142///
143/// # Errors
144///
145/// `22P03` for a value that does not have this form, and `08P01` for a value with too few bytes.
146pub fn snapshot_recv(recv: &mut Recv<'_>) -> Result<Snapshot, TypeError> {
147    let bad = || {
148        TypeError::new(
149            SqlState::INVALID_BINARY_REPRESENTATION,
150            "invalid external pg_snapshot data".to_string(),
151        )
152    };
153    let Some(nxip) = usize::try_from(recv.i32()?).ok().filter(|&n| n <= MAX_NXIP) else {
154        return Err(bad());
155    };
156    let xmin = recv.i64()?.cast_unsigned();
157    let xmax = recv.i64()?.cast_unsigned();
158    if xmin == 0 || xmax == 0 || xmax < xmin {
159        return Err(bad());
160    }
161    let mut xip = Vec::new();
162    let mut last = 0;
163    for _ in 0..nxip {
164        let xid = recv.i64()?.cast_unsigned();
165        if xid < last || xid < xmin || xmax < xid {
166            return Err(bad());
167        }
168        if xid != last {
169            xip.push(xid);
170        }
171        last = xid;
172    }
173    Ok(Snapshot { xmin, xmax, xip })
174}
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179
180    fn text(s: &str) -> String {
181        let mut out = Vec::new();
182        snapshot_out(&snapshot_in(s).unwrap(), &mut out);
183        String::from_utf8(out).unwrap()
184    }
185
186    /// The values and the errors of PostgreSQL 19 for the same text.
187    #[test]
188    fn text_forms() {
189        assert_eq!(text("10:20:10,14,15"), "10:20:10,14,15");
190        assert_eq!(text("10:20:10,10,14"), "10:20:10,14");
191        assert_eq!(text("10:20:10,"), "10:20:10");
192        assert_eq!(text(" 10:20:"), "10:20:");
193        assert_eq!(text("00010:020:"), "10:20:");
194        assert_eq!(text("10:10:"), "10:10:");
195        assert_eq!(
196            text("18446744073709551616:18446744073709551616:"),
197            "18446744073709551615:18446744073709551615:"
198        );
199        for bad in [
200            "20:10:",
201            "10:20:9",
202            "10:20:20",
203            "10:20:14,12",
204            "0:20:",
205            "a:20:",
206            "10:20: ",
207            "10 :20:",
208            "10:20",
209            "-1:20:",
210        ] {
211            let error = snapshot_in(bad).unwrap_err();
212            assert_eq!(
213                error.message,
214                format!("invalid input syntax for type pg_snapshot: \"{bad}\"")
215            );
216        }
217    }
218
219    #[test]
220    fn binary_forms() {
221        let snapshot = snapshot_in("10:20:10,14,15").unwrap();
222        let bytes = snapshot.to_bytes();
223        assert_eq!(bytes.len(), 44);
224        assert_eq!(&bytes[..4], &[0, 0, 0, 3]);
225        assert_eq!(snapshot_recv(&mut Recv::new(&bytes)).unwrap(), snapshot);
226        assert_eq!(Snapshot::from_bytes(&bytes).unwrap(), snapshot);
227        let mut bad = bytes.clone();
228        bad[4..12].copy_from_slice(&30u64.to_be_bytes());
229        assert_eq!(
230            snapshot_recv(&mut Recv::new(&bad)).unwrap_err().message,
231            "invalid external pg_snapshot data"
232        );
233    }
234
235    #[test]
236    fn visibility() {
237        let snapshot = snapshot_in("10:20:10,14,15").unwrap();
238        let visible: Vec<bool> =
239            [9, 10, 11, 14, 19, 20].iter().map(|&xid| snapshot.is_visible(xid)).collect();
240        assert_eq!(visible, [true, false, true, false, true, false]);
241    }
242}
rupg_types/snapshot.rs

rupg_types/
snapshot.rs
