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

lfsx_server/storage/
rewrite.rs
