lfsx_server/storage/
mod.rs

1mod backend;
2mod codec;
3pub mod crypt;
4mod dedupe;
5mod rewrite;
6pub mod s3;
7mod staging;
8mod sweep;
9mod usage;
10mod verify;
11mod walk;
12
13#[cfg(test)]
14mod tests;
15
16use std::path::{Path, PathBuf};
17use std::sync::atomic::{AtomicU64, Ordering};
18use std::time::Instant;
19
20use tokio::sync::Mutex;
21
22use futures_util::{Stream, StreamExt};
23use sha2::{Digest, Sha256};
24use tokio::fs;
25use tokio::io::AsyncWriteExt;
26
27use crate::error::Error;
28use crate::namespace::Namespace;
29
30pub use backend::Store;
31pub use dedupe::DedupeReport;
32use dedupe::shares_bytes_with;
33pub use rewrite::CompressReport;
34pub use verify::VerifyReport;
35
36enum Sink {
37    Raw(fs::File),
38    Framed(Box<codec::Writer>),
39}
40
41impl Sink {
42    async fn write(&mut self, chunk: &[u8]) -> Result<(), Error> {
43        match self {
44            Self::Raw(file) => Ok(file.write_all(chunk).await?),
45            Self::Framed(writer) => writer.push(chunk).await,
46        }
47    }
48
49    async fn finish(self) -> Result<(), Error> {
50        match self {
51            Self::Raw(mut file) => {
52                file.flush().await?;
53                Ok(file.sync_all().await?)
54            }
55            Self::Framed(writer) => writer.finish().await,
56        }
57    }
58}
59
60// A transfer that has passed every check and is waiting to be put somewhere.
61pub struct Staged {
62    pub path: PathBuf,
63    destination: PathBuf,
64    pub written: u64,
65    fresh: bool,
66}
67
68// What a download reads from, whether or not the bytes on disk are the object.
69pub enum Object {
70    Raw {
71        file: fs::File,
72        size: u64,
73    },
74    Framed(codec::Framed),
75    Remote {
76        bucket: s3::S3Store,
77        oid: String,
78        size: u64,
79    },
80}
81
82impl Object {
83    pub fn size(&self) -> u64 {
84        match self {
85            Self::Raw { size, .. } => *size,
86            Self::Framed(framed) => framed.plaintext(),
87            Self::Remote { size, .. } => *size,
88        }
89    }
90
91    pub async fn stream(
92        self,
93        start: u64,
94        length: u64,
95    ) -> Result<futures_util::stream::BoxStream<'static, Result<axum::body::Bytes, Error>>, Error>
96    {
97        use futures_util::StreamExt;
98        use tokio::io::AsyncSeekExt;
99
100        match self {
101            Self::Raw { mut file, .. } => {
102                file.seek(std::io::SeekFrom::Start(start)).await?;
103                let reader =
104                    tokio_util::io::ReaderStream::new(tokio::io::AsyncReadExt::take(file, length));
105
106                Ok(reader.map(|chunk| chunk.map_err(Error::from)).boxed())
107            }
108            Self::Framed(framed) => Ok(framed.stream(start, length).boxed()),
109            Self::Remote { bucket, oid, .. } => {
110                let chunks = bucket.read(&oid, start, length).await?;
111
112                Ok(chunks
113                    .map(|chunk| {
114                        chunk.map_err(|error| Error::Storage(std::io::Error::other(error)))
115                    })
116                    .boxed())
117            }
118        }
119    }
120}
121pub use staging::{Reclaimed, reclaim};
122pub use sweep::SweepReport;
123
124// What is left of a repository's budget for one transfer. It travels with the
125// upload because a client that skips negotiation may also skip declaring a
126// size, and a budget checked once against a number the client chose is not a
127// budget.
128#[derive(Debug, Clone, Copy)]
129pub struct Budget {
130    pub used: u64,
131    pub limit: u64,
132}
133
134impl Budget {
135    pub fn exceeded_by(&self, arriving: u64) -> bool {
136        self.used + arriving > self.limit
137    }
138
139    pub fn refusal(&self) -> Error {
140        Error::OverQuota {
141            used: self.used,
142            limit: self.limit,
143        }
144    }
145}
146
147// What a write did, which the caller needs and the byte count alone does not
148// say: an object the repository already held costs it no room, so counting it
149// again would push a repository over a quota it never grew into.
150#[derive(Debug)]
151pub struct Written {
152    pub bytes: u64,
153    pub fresh: bool,
154}
155
156pub struct LocalStore {
157    root: PathBuf,
158    counter: AtomicU64,
159    usage: Mutex<Option<(Instant, u64, u64)>>,
160    scans: AtomicU64,
161    max_object_size: Option<u64>,
162    compression: Option<i32>,
163    // Shared rather than cloned: the staging store and the store proper are two
164    // handles onto one deployment's keys.
165    keys: Option<std::sync::Arc<crypt::Keyring>>,
166}
167
168impl LocalStore {
169    pub fn new(root: impl Into<PathBuf>) -> Self {
170        Self {
171            root: root.into(),
172            counter: AtomicU64::new(0),
173            usage: Mutex::new(None),
174            scans: AtomicU64::new(0),
175            max_object_size: None,
176            compression: None,
177            keys: None,
178        }
179    }
180
181    pub fn with_compression(mut self, level: Option<i32>) -> Self {
182        self.compression = level;
183        self
184    }
185
186    pub fn with_max_object_size(mut self, limit: Option<u64>) -> Self {
187        self.max_object_size = limit;
188        self
189    }
190
191    pub fn with_encryption(mut self, keys: Option<std::sync::Arc<crypt::Keyring>>) -> Self {
192        self.keys = keys;
193        self
194    }
195
196    pub(super) fn keyring(&self) -> Option<&std::sync::Arc<crypt::Keyring>> {
197        self.keys.as_ref()
198    }
199
200    pub fn encrypts(&self) -> bool {
201        self.keys.is_some()
202    }
203
204    // Whether an object in this store is a framing of the bytes rather than the
205    // bytes. Compression and encryption both put a frame under the plaintext
206    // digest, and only the codec turns one back into the object a client asked
207    // for by that name.
208    pub fn frames(&self) -> bool {
209        self.compression.is_some() || self.keys.is_some()
210    }
211
212    pub fn validate_oid(oid: &str) -> Result<(), Error> {
213        let well_formed = oid.len() == 64
214            && oid
215                .bytes()
216                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b));
217
218        well_formed.then_some(()).ok_or(Error::MalformedOid)
219    }
220
221    fn object_path(&self, ns: &Namespace, oid: &str) -> PathBuf {
222        self.root
223            .join(ns.org())
224            .join(ns.repo())
225            .join(&oid[0..2])
226            .join(&oid[2..4])
227            .join(oid)
228    }
229
230    fn content_path(&self, oid: &str) -> PathBuf {
231        self.root
232            .join(".content")
233            .join(&oid[0..2])
234            .join(&oid[2..4])
235            .join(oid)
236    }
237
238    pub fn scans(&self) -> u64 {
239        self.scans.load(Ordering::Relaxed)
240    }
241
242    pub async fn writable(&self) -> Result<(), Error> {
243        fs::create_dir_all(&self.root).await?;
244
245        let ticket = self.counter.fetch_add(1, Ordering::Relaxed);
246        let probe = self.root.join(format!(".readiness.{ticket}"));
247
248        fs::write(&probe, b"").await?;
249        fs::remove_file(&probe).await?;
250
251        Ok(())
252    }
253
254    pub async fn exists(&self, ns: &Namespace, oid: &str) -> bool {
255        Self::validate_oid(oid).is_ok() && fs::metadata(self.object_path(ns, oid)).await.is_ok()
256    }
257
258    pub async fn open(&self, ns: &Namespace, oid: &str) -> Result<Object, Error> {
259        Self::validate_oid(oid)?;
260        let path = self.object_path(ns, oid);
261        let file = fs::File::open(&path).await.map_err(|_| Error::NotFound)?;
262        let on_disk = file.metadata().await?.len();
263
264        match codec::Framed::open(
265            codec::Reader::File(file),
266            on_disk,
267            self.keys.as_deref(),
268            oid,
269        )
270        .await?
271        {
272            Some(framed) => Ok(Object::Framed(framed)),
273            None => Ok(Object::Raw {
274                file: fs::File::open(&path).await.map_err(|_| Error::NotFound)?,
275                size: on_disk,
276            }),
277        }
278    }
279
280    // Everything a transfer has to survive before it counts as an object: the
281    // digest it claims, the size it declared, the ceiling on a single object and
282    // the repository's remaining budget. It ends on local disk whatever the
283    // backend is, because a bucket cannot be asked to hold bytes that might turn
284    // out to be the wrong ones.
285    pub async fn stage<S, E>(
286        &self,
287        ns: &Namespace,
288        oid: &str,
289        expected_size: Option<u64>,
290        budget: Option<Budget>,
291        mut chunks: S,
292    ) -> Result<Staged, Error>
293    where
294        S: Stream<Item = Result<axum::body::Bytes, E>> + Unpin,
295        E: std::error::Error + Send + Sync + 'static,
296    {
297        Self::validate_oid(oid)?;
298
299        if let Some(limit) = self.max_object_size
300            && expected_size.is_some_and(|declared| declared > limit)
301        {
302            return Err(Error::TooLarge { limit });
303        }
304
305        let path = self.object_path(ns, oid);
306        let parent = path.parent().expect("object paths always have a parent");
307        fs::create_dir_all(parent).await?;
308
309        // A retried transfer of an object this repository already holds costs it
310        // no room, so it must not count against the budget a second time.
311        let fresh = fs::metadata(&path).await.is_err();
312        let staged = self.staging_path(parent, oid);
313
314        match self.stream_to(&staged, oid, budget, &mut chunks).await {
315            Ok((digest, written)) => {
316                if let Err(error) = Self::agrees(oid, expected_size, &digest, written) {
317                    let _ = fs::remove_file(&staged).await;
318                    return Err(error);
319                }
320
321                Ok(Staged {
322                    path: staged,
323                    destination: path,
324                    written,
325                    fresh,
326                })
327            }
328            Err(error) => {
329                let _ = fs::remove_file(&staged).await;
330                Err(error)
331            }
332        }
333    }
334
335    fn agrees(
336        oid: &str,
337        expected_size: Option<u64>,
338        digest: &str,
339        written: u64,
340    ) -> Result<(), Error> {
341        if let Some(declared) = expected_size.filter(|declared| *declared != written) {
342            return Err(Error::SizeMismatch {
343                declared,
344                actual: written,
345            });
346        }
347
348        if digest != oid {
349            return Err(Error::OidMismatch {
350                declared: oid.to_owned(),
351                actual: digest.to_owned(),
352            });
353        }
354
355        Ok(())
356    }
357
358    pub async fn write<S, E>(
359        &self,
360        ns: &Namespace,
361        oid: &str,
362        expected_size: Option<u64>,
363        budget: Option<Budget>,
364        chunks: S,
365    ) -> Result<Written, Error>
366    where
367        S: Stream<Item = Result<axum::body::Bytes, E>> + Unpin,
368        E: std::error::Error + Send + Sync + 'static,
369    {
370        let staged = self.stage(ns, oid, expected_size, budget, chunks).await?;
371
372        self.link_or_move(&staged.path, &staged.destination, oid)
373            .await?;
374
375        Ok(Written {
376            bytes: staged.written,
377            fresh: staged.fresh,
378        })
379    }
380
381    fn staging_path(&self, parent: &Path, oid: &str) -> PathBuf {
382        let ticket = self.counter.fetch_add(1, Ordering::Relaxed);
383        parent.join(format!("{oid}.{ticket}.part"))
384    }
385
386    async fn stream_to<S, E>(
387        &self,
388        staged: &Path,
389        oid: &str,
390        budget: Option<Budget>,
391        chunks: &mut S,
392    ) -> Result<(String, u64), Error>
393    where
394        S: Stream<Item = Result<axum::body::Bytes, E>> + Unpin,
395        E: std::error::Error + Send + Sync + 'static,
396    {
397        let file = fs::File::create(staged).await?;
398        // The digest, the declared size and the budget are all counted on the
399        // plaintext going past, whatever the bytes look like once they land —
400        // so compression is a different sink, not a different path.
401        let mut sink = match (self.compression, self.keys.as_deref()) {
402            (None, None) => Sink::Raw(file),
403            (level, keys) => Sink::Framed(Box::new(
404                codec::Writer::open(file, level, keys.map(crypt::Keyring::writing), oid).await?,
405            )),
406        };
407        let mut hasher = Sha256::new();
408        let mut written = 0u64;
409
410        while let Some(chunk) = chunks.next().await {
411            let chunk = chunk.map_err(std::io::Error::other)?;
412            hasher.update(&chunk);
413            written += chunk.len() as u64;
414
415            // The declared size is a claim by the client, so the ceiling has to
416            // hold against a body that ignores it. Stopping at the chunk that
417            // crosses the line is the point: reading to the end to find out how
418            // big it was would be the outage this limit exists to prevent.
419            if let Some(limit) = self.max_object_size.filter(|limit| written > *limit) {
420                return Err(Error::TooLarge { limit });
421            }
422
423            if let Some(budget) = budget.filter(|budget| budget.exceeded_by(written)) {
424                return Err(budget.refusal());
425            }
426
427            sink.write(&chunk).await?;
428        }
429
430        sink.finish().await?;
431
432        Ok((hex::encode(hasher.finalize()), written))
433    }
434
435    // One copy of the bytes under .content, and a hard link per repository that
436    // holds them. Two projects sharing an asset pack cost the disk once, and the
437    // link count is the reference count — the filesystem does the bookkeeping, so
438    // nothing can leak a repository's contents to another and nothing needs a
439    // migration: objects already sitting at their repository path keep working as
440    // ordinary files with one link.
441    async fn link_or_move(&self, staged: &Path, final_path: &Path, oid: &str) -> Result<(), Error> {
442        let content = self.content_path(oid);
443        let parent = content.parent().expect("content paths have a parent");
444        fs::create_dir_all(parent).await?;
445
446        if fs::metadata(&content).await.is_err() {
447            fs::rename(staged, &content).await?;
448        }
449
450        match self.link(&content, final_path).await {
451            // The content was collected between finding it and linking to it:
452            // a concurrent retain on another repository dropped its last other
453            // reference. The staged copy is still here precisely for this, so
454            // put it back and link again rather than failing a push that did
455            // nothing wrong.
456            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
457                fs::rename(staged, &content).await?;
458                self.link(&content, final_path).await?;
459            }
460            outcome => outcome?,
461        }
462
463        let _ = fs::remove_file(staged).await;
464        Ok(())
465    }
466
467    async fn link(&self, content: &Path, final_path: &Path) -> Result<(), std::io::Error> {
468        let from = content.to_path_buf();
469        let to = final_path.to_path_buf();
470        let linked = tokio::task::spawn_blocking(move || std::fs::hard_link(&from, &to))
471            .await
472            .map_err(std::io::Error::other)?;
473
474        match linked {
475            Ok(()) => Ok(()),
476            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(()),
477            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Err(error),
478            // A filesystem without hard links, or one crossing a device
479            // boundary: fall back to a full copy so the transfer still
480            // succeeds. The disk pays for it, the client never notices.
481            Err(_) => fs::copy(content, final_path).await.map(|_| ()),
482        }
483    }
484}
lfsx_server/storage/mod.rs

lfsx_server/storage/
mod.rs
