lfsx_server/storage/
rewrite.rs

1use std::path::Path;
2
3use serde::Serialize;
4use sha2::{Digest, Sha256};
5use tokio::fs;
6use tokio::io::AsyncReadExt;
7
8use super::{LocalStore, shares_bytes_with};
9use super::{codec, crypt};
10use crate::error::Error;
11use crate::namespace::Namespace;
12
13#[derive(Debug, Default, Serialize, PartialEq, Eq)]
14pub struct CompressReport {
15    pub inspected: u64,
16    pub compressed: u64,
17    pub already: u64,
18    pub left_alone: u64,
19    pub refused: u64,
20    pub before: u64,
21    pub after: u64,
22    pub incomplete: bool,
23    pub dry_run: bool,
24}
25
26impl LocalStore {
27    // Turning compression on only changes what arrives next. This is how a store
28    // that predates it stops paying full price for what it already holds.
29    pub async fn compress(&self, ns: &Namespace, dry_run: bool) -> Result<CompressReport, Error> {
30        let level = self.compression.ok_or(Error::CompressionDisabled)?;
31
32        let walk = self.objects_of(ns).await;
33        let mut report = CompressReport {
34            dry_run,
35            incomplete: !walk.complete,
36            ..CompressReport::default()
37        };
38
39        for found in walk.objects {
40            report.inspected += 1;
41            let on_disk = fs::metadata(&found.path).await?.len();
42            report.before += on_disk;
43
44            if self.is_framed(&found.path, &found.oid, on_disk).await? {
45                report.already += 1;
46                report.after += on_disk;
47                continue;
48            }
49
50            report.after += self
51                .compress_object(&found.path, &found.oid, on_disk, level, &mut report)
52                .await?;
53        }
54
55        if !dry_run && report.compressed > 0 {
56            self.forget(ns).await;
57        }
58
59        Ok(report)
60    }
61
62    async fn is_framed(&self, path: &Path, oid: &str, on_disk: u64) -> Result<bool, Error> {
63        let file = fs::File::open(path).await?;
64
65        Ok(
66            codec::Framed::open(file, on_disk, self.keys.as_deref(), oid)
67                .await?
68                .is_some(),
69        )
70    }
71
72    async fn compress_object(
73        &self,
74        path: &Path,
75        oid: &str,
76        on_disk: u64,
77        level: i32,
78        report: &mut CompressReport,
79    ) -> Result<u64, Error> {
80        let parent = path.parent().expect("objects live in a fanout directory");
81        let staged = self.staging_path(parent, oid);
82
83        let (digest, compressed) = match self.rewrite(path, &staged, oid, level).await {
84            Ok(measured) => measured,
85            Err(error) => {
86                let _ = fs::remove_file(&staged).await;
87                return Err(error);
88            }
89        };
90
91        // The name is the digest, so this is the same check an operator would
92        // run by hand — and the last chance to run it, since afterwards the file
93        // is no longer the bytes it is named after.
94        if digest != oid {
95            tracing::warn!(oid, %digest, "object does not hash to its own name, leaving it alone");
96            let _ = fs::remove_file(&staged).await;
97            report.refused += 1;
98            return Ok(on_disk);
99        }
100
101        // An object that will not compress costs a header and an index to store
102        // this way. Leaving it is not a failure, it is the right answer.
103        if compressed >= on_disk || report.dry_run {
104            let _ = fs::remove_file(&staged).await;
105            if compressed >= on_disk {
106                report.left_alone += 1;
107                return Ok(on_disk);
108            }
109
110            report.compressed += 1;
111            return Ok(compressed);
112        }
113
114        self.swap_in(path, &staged, oid).await?;
115        report.compressed += 1;
116
117        Ok(compressed)
118    }
119
120    // Every repository holding these bytes has a link to one file, so replacing
121    // this repository's link with a compressed copy would break the sharing the
122    // deduplication just built. The shared copy is what gets replaced, and this
123    // repository is relinked to it. Repositories that have not run this yet keep
124    // the old bytes alive through their own links until their turn.
125    async fn swap_in(&self, path: &Path, staged: &Path, oid: &str) -> Result<(), Error> {
126        let content = self.content_path(oid);
127
128        if !shares_bytes_with(path, &content).await {
129            return Ok(fs::rename(staged, path).await?);
130        }
131
132        fs::rename(staged, &content).await?;
133
134        let parent = path.parent().expect("objects live in a fanout directory");
135        let relink = self.staging_path(parent, oid);
136        self.link(&content, &relink).await?;
137
138        Ok(fs::rename(&relink, path).await?)
139    }
140
141    async fn rewrite(
142        &self,
143        path: &Path,
144        staged: &Path,
145        oid: &str,
146        level: i32,
147    ) -> Result<(String, u64), Error> {
148        let mut source = fs::File::open(path).await?;
149        let mut writer = codec::Writer::open(
150            fs::File::create(staged).await?,
151            Some(level),
152            self.keys.as_deref().map(crypt::Keyring::writing),
153            oid,
154        )
155        .await?;
156        let mut hasher = Sha256::new();
157        let mut buffer = vec![0u8; 1024 * 1024];
158
159        loop {
160            let read = source.read(&mut buffer).await?;
161            if read == 0 {
162                break;
163            }
164
165            hasher.update(&buffer[..read]);
166            writer.push(&buffer[..read]).await?;
167        }
168
169        writer.finish().await?;
170
171        Ok((
172            hex::encode(hasher.finalize()),
173            fs::metadata(staged).await?.len(),
174        ))
175    }
176}
177
178#[cfg(test)]
179mod tests;
lfsx_server/storage/rewrite.rs

lfsx_server/storage/
rewrite.rs
