vcon_server/store/
export.rs

1//! What `export parquet` reads of one tenant's Store: every kept version in
2//! one snapshot of the catalog, then each version's object, outside any
3//! transaction so the export never holds back the server's checkpoints.
4//! Reading changes nothing.
5
6use uuid::Uuid;
7
8use super::packed::Unread;
9use super::{Store, StoreError, key_failed, read_failed};
10use crate::backend::ObjectError;
11use crate::envelope::{self, Context};
12use crate::ids::{TenantId, VersionId};
13
14/// A kept version as the export lists it.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct KeptVersion {
17    pub version: VersionId,
18    pub uuid: Uuid,
19    /// The form's name as stored: unsigned, signed or encrypted.
20    pub form: String,
21    pub gzip: bool,
22    /// The body's size in bytes, as received.
23    pub size: i64,
24    /// Nanoseconds since the epoch.
25    pub stored_at: i64,
26    /// Whether it is its uuid's current version and the uuid is not
27    /// deleted: what a read returns.
28    pub live: bool,
29    /// The data key, wrapped under the tenant key.
30    pub wrapped: Vec<u8>,
31    /// Where its sealed blob is when it is packed; None: in its object.
32    pub(crate) place: Option<crate::backend::Packed>,
33}
34
35impl Store {
36    /// The tenant this Store holds.
37    pub fn tenant(&self) -> TenantId {
38        self.tenant
39    }
40
41    /// Every kept version, live or hidden, stored at or after `since` and
42    /// before `until` (nanoseconds since the epoch), by store time and then
43    /// version id, all from one snapshot of the catalog.
44    pub fn kept_versions(
45        &self,
46        since: Option<i64>,
47        until: Option<i64>,
48    ) -> Result<Vec<KeptVersion>, StoreError> {
49        let rows = self.catalog.snapshot()?.exported(since, until)?;
50        Ok(rows
51            .into_iter()
52            .map(|row| KeptVersion {
53                version: row.version,
54                uuid: row.uuid,
55                form: row.form,
56                gzip: row.gzip,
57                size: row.size,
58                stored_at: row.stored_at,
59                live: row.live,
60                wrapped: row.wrapped,
61                place: row.place,
62            })
63            .collect())
64    }
65
66    /// The sealed blob of a version `kept_versions` listed, from its object
67    /// or its pack file, byte for byte. None when the version has been
68    /// purged since: its row is gone and its blob with it. A blob missing
69    /// while its row is still there is damage.
70    pub fn sealed_of(&self, kept: &KeptVersion) -> Result<Option<Vec<u8>>, StoreError> {
71        let mut place = kept.place.clone();
72        loop {
73            match self.sealed_blob(kept.version, place.as_ref()) {
74                Ok(sealed) => return Ok(Some(sealed)),
75                Err(Unread::Damaged) => return Err(StoreError::Damaged),
76                Err(Unread::Object(ObjectError::Missing)) => {
77                    // A purge removes the row first, then the object; a
78                    // purge's rewrite moves a packed row to a new file
79                    // first, then deletes the old one.
80                    let row = self.catalog.snapshot()?.version_row(kept.version)?;
81                    match row {
82                        None => return Ok(None),
83                        Some(row) if row.place != place => place = row.place,
84                        Some(_) => return Err(StoreError::Damaged),
85                    }
86                }
87                Err(Unread::Object(e @ ObjectError::Io(_))) => return Err(read_failed(e)),
88            }
89        }
90    }
91
92    /// The body `sealed`, the object of `kept`, holds: what `get` returns.
93    pub fn open_sealed(&self, kept: &KeptVersion, sealed: &[u8]) -> Result<Vec<u8>, StoreError> {
94        let ctx = Context::Version {
95            tenant: self.tenant,
96            uuid: kept.uuid,
97            version: kept.version,
98        };
99        let data_key =
100            envelope::unwrap(&self.tenant_key, &ctx, &kept.wrapped).map_err(key_failed)?;
101        envelope::open(&data_key, &ctx, sealed).map_err(|_| StoreError::Damaged)
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use time::Duration;
108
109    use super::*;
110    use crate::audit::Actor;
111    use crate::clock::{Clock, nanos};
112    use crate::fs::Fs;
113    use crate::retention::Retention;
114    use crate::store::Created;
115    use crate::store::testing::vcon;
116    use crate::store::tests::{Capture, Fixture, ReadsFail, fixture};
117
118    const A: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc46";
119    const B: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc47";
120    const C: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc48";
121    const D: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc49";
122
123    fn later(f: &Fixture, by: Duration) {
124        f.clock.set(f.clock.now() + by);
125    }
126
127    fn create(f: &Fixture, uuid: &str, subject: &str) -> VersionId {
128        let (found, body) = vcon(uuid, subject);
129        match f.store.create(&f.caller, &found, &body).expect("create") {
130            Created::New(v) => v,
131            other => panic!("{other:?}"),
132        }
133    }
134
135    /// A's first version replaced by its second, B stored, C deleted, D
136    /// purged; a minute between each store. Returns A's two versions, B's
137    /// and C's, in that order.
138    fn stored(f: &Fixture) -> [VersionId; 4] {
139        let a1 = create(f, A, "first");
140        later(f, Duration::minutes(1));
141        let (found, body) = vcon(A, "second");
142        let a2 = f
143            .store
144            .replace(&f.caller, a1, &found, &body)
145            .expect("replace");
146        later(f, Duration::minutes(1));
147        let b1 = create(f, B, "other");
148        later(f, Duration::minutes(1));
149        let c1 = create(f, C, "deleted");
150        f.store
151            .delete(&f.caller, C.parse().expect("uuid"), Retention::Years(7))
152            .expect("delete");
153        later(f, Duration::minutes(1));
154        create(f, D, "purged");
155        // A delete under a period already passed purges at once.
156        f.store
157            .delete(&f.caller, D.parse().expect("uuid"), Retention::Days(0))
158            .expect("delete");
159        assert_eq!(
160            f.store.legal_get(&f.caller, D.parse().expect("uuid"), None),
161            Err(StoreError::Gone),
162            "D is purged"
163        );
164        later(f, Duration::minutes(1));
165        [a1, a2, b1, c1]
166    }
167
168    #[test]
169    fn every_kept_version_is_listed_once_and_no_purged_one() {
170        let f = fixture();
171        let [a1, a2, b1, c1] = stored(&f);
172        let kept = f.store.kept_versions(None, None).expect("lists");
173        let listed: Vec<(VersionId, bool)> = kept.iter().map(|k| (k.version, k.live)).collect();
174        assert_eq!(
175            listed,
176            [(a1, false), (a2, true), (b1, true), (c1, false)],
177            "in store order; only a current version of a live uuid is live"
178        );
179        let a = &kept[0];
180        assert_eq!(a.uuid, A.parse::<Uuid>().expect("uuid"), "its uuid");
181        assert_eq!(a.form, "unsigned", "its form");
182        assert!(!a.gzip, "not gzipped");
183        assert_eq!(a.size, vcon(A, "first").1.len() as i64, "the body's size");
184        assert!(!a.wrapped.is_empty(), "its wrapped key");
185        assert!(
186            kept.windows(2).all(|w| w[0].stored_at < w[1].stored_at),
187            "store times rise"
188        );
189    }
190
191    /// `since` is inclusive and `until` exclusive, as for the list.
192    #[test]
193    fn the_window_takes_since_and_leaves_until() {
194        let f = fixture();
195        let [_, a2, b1, _] = stored(&f);
196        let all = f.store.kept_versions(None, None).expect("lists");
197        let at = |v: VersionId| all.iter().find(|k| k.version == v).expect("kept").stored_at;
198        let window = f
199            .store
200            .kept_versions(Some(at(a2)), Some(at(b1) + 1))
201            .expect("lists");
202        let listed: Vec<VersionId> = window.iter().map(|k| k.version).collect();
203        assert_eq!(listed, [a2, b1], "from A2's time to just after B1's");
204        let none = f
205            .store
206            .kept_versions(Some(at(a2)), Some(at(a2)))
207            .expect("lists");
208        assert!(none.is_empty(), "until is left out: {none:?}");
209        assert!(
210            f.store
211                .kept_versions(Some(nanos(f.clock.now())), None)
212                .expect("lists")
213                .is_empty(),
214            "nothing stored since the last store"
215        );
216    }
217
218    #[test]
219    fn a_sealed_object_opens_to_the_body_get_returns() {
220        let f = fixture();
221        stored(&f);
222        for kept in f.store.kept_versions(None, None).expect("lists") {
223            let sealed = f.store.sealed_of(&kept).expect("reads").expect("kept");
224            let body = f.store.open_sealed(&kept, &sealed).expect("opens");
225            let read = f
226                .store
227                .legal_get(&f.caller, kept.uuid, Some(kept.version))
228                .expect("legal read");
229            assert_eq!(
230                body, read.stored.body,
231                "{}: the body as stored",
232                kept.version
233            );
234            let mut other = kept.clone();
235            other.version = VersionId([0xee; 16]);
236            assert_eq!(
237                f.store.open_sealed(&other, &sealed),
238                Err(StoreError::Damaged),
239                "bound to its version"
240            );
241        }
242    }
243
244    /// An object the storage fails to give is Io, logged by its kind as
245    /// every read is: the export stops, and calls no version damaged.
246    #[test]
247    fn an_object_the_storage_fails_to_give_is_io_and_logged() {
248        let mut f = fixture();
249        stored(&f);
250        let kept = f.store.kept_versions(None, None).expect("lists");
251        f.store.objects = std::sync::Arc::new(ReadsFail {
252            inner: f.store.objects.clone(),
253            fail: || ObjectError::Io(std::io::Error::from_raw_os_error(libc::EIO)),
254        });
255        let capture = Capture::default();
256        let subscriber = tracing_subscriber::fmt()
257            .with_writer(capture.clone())
258            .with_ansi(false)
259            .finish();
260        let read = tracing::subscriber::with_default(subscriber, || f.store.sealed_of(&kept[0]));
261        assert_eq!(read, Err(StoreError::Io), "a storage failure is Io");
262        let log = String::from_utf8(capture.0.lock().expect("lock").clone()).expect("UTF-8");
263        assert!(
264            log.contains("a stored version's object could not be read")
265                && log.contains(&format!("os_error={}", libc::EIO)),
266            "the failure is logged by its kind: {log}"
267        );
268    }
269
270    /// A version purged after the listing is left out; an object lost
271    /// while its row stays is damage, never left out.
272    #[test]
273    fn a_purge_after_the_listing_is_skipped_and_a_lost_object_is_damage() {
274        let f = fixture();
275        let [_, _, b1, c1] = stored(&f);
276        let kept = f.store.kept_versions(None, None).expect("lists");
277        let of = |v: VersionId| kept.iter().find(|k| k.version == v).expect("kept");
278        f.store
279            .purge_hidden(
280                Actor::Cli,
281                Retention::Days(0),
282                Some(C.parse().expect("uuid")),
283            )
284            .expect("purges C");
285        assert_eq!(
286            f.store.sealed_of(of(c1)),
287            Ok(None),
288            "C1 was purged since the listing"
289        );
290        f.fs.remove(&f.store.objects_dir.join(b1.to_string()))
291            .expect("removes B1's object");
292        assert_eq!(
293            f.store.sealed_of(of(b1)),
294            Err(StoreError::Damaged),
295            "B1's row is still there"
296        );
297    }
298}
vcon_server/store/export.rs

vcon_server/store/
export.rs
