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(codec::Framed::open(
66            codec::Reader::File(file),
67            on_disk,
68            self.keys.as_deref(),
69            oid,
70        )
71        .await?
72        .is_some())
73    }
74
75    async fn compress_object(
76        &self,
77        path: &Path,
78        oid: &str,
79        on_disk: u64,
80        level: i32,
81        report: &mut CompressReport,
82    ) -> Result<u64, Error> {
83        let parent = path.parent().expect("objects live in a fanout directory");
84        let staged = self.staging_path(parent, oid);
85
86        let (digest, compressed) = match self.rewrite(path, &staged, oid, level).await {
87            Ok(measured) => measured,
88            Err(error) => {
89                let _ = fs::remove_file(&staged).await;
90                return Err(error);
91            }
92        };
93
94        // The name is the digest, so this is the same check an operator would
95        // run by hand — and the last chance to run it, since afterwards the file
96        // is no longer the bytes it is named after.
97        if digest != oid {
98            tracing::warn!(oid, %digest, "object does not hash to its own name, leaving it alone");
99            let _ = fs::remove_file(&staged).await;
100            report.refused += 1;
101            return Ok(on_disk);
102        }
103
104        // An object that will not compress costs a header and an index to store
105        // this way. Leaving it is not a failure, it is the right answer.
106        if compressed >= on_disk || report.dry_run {
107            let _ = fs::remove_file(&staged).await;
108            if compressed >= on_disk {
109                report.left_alone += 1;
110                return Ok(on_disk);
111            }
112
113            report.compressed += 1;
114            return Ok(compressed);
115        }
116
117        self.swap_in(path, &staged, oid).await?;
118        report.compressed += 1;
119
120        Ok(compressed)
121    }
122
123    // Every repository holding these bytes has a link to one file, so replacing
124    // this repository's link with a compressed copy would break the sharing the
125    // deduplication just built. The shared copy is what gets replaced, and this
126    // repository is relinked to it. Repositories that have not run this yet keep
127    // the old bytes alive through their own links until their turn.
128    async fn swap_in(&self, path: &Path, staged: &Path, oid: &str) -> Result<(), Error> {
129        let content = self.content_path(oid);
130
131        if !shares_bytes_with(path, &content).await {
132            return Ok(fs::rename(staged, path).await?);
133        }
134
135        fs::rename(staged, &content).await?;
136
137        let parent = path.parent().expect("objects live in a fanout directory");
138        let relink = self.staging_path(parent, oid);
139        self.link(&content, &relink).await?;
140
141        Ok(fs::rename(&relink, path).await?)
142    }
143
144    async fn rewrite(
145        &self,
146        path: &Path,
147        staged: &Path,
148        oid: &str,
149        level: i32,
150    ) -> Result<(String, u64), Error> {
151        let mut source = fs::File::open(path).await?;
152        let mut writer = codec::Writer::open(
153            fs::File::create(staged).await?,
154            Some(level),
155            self.keys.as_deref().map(crypt::Keyring::writing),
156            oid,
157        )
158        .await?;
159        let mut hasher = Sha256::new();
160        let mut buffer = vec![0u8; 1024 * 1024];
161
162        loop {
163            let read = source.read(&mut buffer).await?;
164            if read == 0 {
165                break;
166            }
167
168            hasher.update(&buffer[..read]);
169            writer.push(&buffer[..read]).await?;
170        }
171
172        writer.finish().await?;
173
174        Ok((
175            hex::encode(hasher.finalize()),
176            fs::metadata(staged).await?.len(),
177        ))
178    }
179}
180
181#[cfg(test)]
182mod tests;
lfsx_server/storage/rewrite.rs

lfsx_server/storage/
rewrite.rs
