feat(gitlfs): add gitlfs crate (oid, store, error)
zhenyi committed
Added lib/gitlfs/error.rs lib/gitlfs/error.rs +20 −0 View lib/gitlfs/error.rs
@@ -0,0 +1,20 @@
1//! LFS storage-layer error type.
2
3/// Unified error for the `gitlfs` crate.
4#[derive(Debug, thiserror::Error)]
5pub enum LfsError {
6 #[error("invalid oid: {0}")]
7 InvalidOid(String),
8 #[error("lfs object not found: {0}")]
9 NotFound(String),
10 #[error("lfs object size mismatch: expected {expected}, got {actual}")]
11 SizeMismatch { expected: u64, actual: u64 },
12 #[error("lfs object hash mismatch: expected {expected}, got {actual}")]
13 HashMismatch { expected: String, actual: String },
14 #[error("lfs payload too large: limit {limit} bytes")]
15 PayloadTooLarge { limit: u64 },
16 #[error("io error: {0}")]
17 Io(String),
18 #[error("storage error: {0}")]
19 Storage(#[from] storage::StorageError),
20}
No newline at end of file
Added lib/gitlfs/oid.rs lib/gitlfs/oid.rs +99 −0 View lib/gitlfs/oid.rs
@@ -0,0 +1,99 @@
1//! LFS object identifier: a validated SHA-256 hex digest.
2
3use sha2::{Digest, Sha256};
4
5use crate::error::LfsError;
6
7const HEX_CHARS: usize = 64;
8const HEX: &[u8; 16] = b"0123456789abcdef";
9
10fn hex_encode(data: &[u8]) -> String {
11 let mut out = String::with_capacity(data.len() * 2);
12 for &b in data {
13 out.push(HEX[(b >> 4) as usize] as char);
14 out.push(HEX[(b & 0x0f) as usize] as char);
15 }
16 out
17}
18
19/// A validated LFS object id (SHA-256, 64 lowercase hex chars).
20#[derive(Debug, Clone, PartialEq, Eq, Hash)]
21pub struct LfsOid(String);
22
23impl LfsOid {
24 /// Validate a hex string and build an [`LfsOid`].
25 pub fn new(hex: impl AsRef<str>) -> Result<Self, LfsError> {
26 let hex = hex.as_ref();
27 if hex.len() != HEX_CHARS || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
28 return Err(LfsError::InvalidOid(hex.to_string()));
29 }
30 Ok(Self(hex.to_ascii_lowercase()))
31 }
32
33 /// Compute the SHA-256 oid of `data`.
34 pub fn compute(data: &[u8]) -> Self {
35 Self(hex_encode(&Sha256::digest(data)))
36 }
37
38 /// Build an [`LfsOid`] from an in-progress SHA-256 hasher.
39 pub fn from_hasher(hasher: Sha256) -> Self {
40 Self(hex_encode(&hasher.finalize()))
41 }
42
43 /// The 64-char lowercase hex representation.
44 pub fn as_str(&self) -> &str {
45 &self.0
46 }
47}
48
49impl std::fmt::Display for LfsOid {
50 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51 f.write_str(&self.0)
52 }
53}
54
55impl AsRef<str> for LfsOid {
56 fn as_ref(&self) -> &str {
57 &self.0
58 }
59}
60
61#[cfg(test)]
62mod tests {
63 use super::*;
64
65 #[test]
66 fn oid_new_accepts_64_hex() {
67 let hex = "0123456789abcdef".repeat(4);
68 assert!(LfsOid::new(&hex).is_ok());
69 }
70
71 #[test]
72 fn oid_new_normalizes_uppercase() {
73 let oid = LfsOid::new(&"A".repeat(64)).unwrap();
74 assert_eq!(oid.as_str(), &"a".repeat(64));
75 }
76
77 #[test]
78 fn oid_new_rejects_bad_length() {
79 assert!(LfsOid::new("").is_err());
80 assert!(LfsOid::new("abc").is_err());
81 assert!(LfsOid::new(&"a".repeat(63)).is_err());
82 assert!(LfsOid::new(&"a".repeat(65)).is_err());
83 }
84
85 #[test]
86 fn oid_new_rejects_non_hex() {
87 assert!(LfsOid::new(&"z".repeat(64)).is_err());
88 }
89
90 #[test]
91 fn oid_compute_matches_sha256() {
92 // sha256(b"hello\n") = 5891b5b5...
93 let oid = LfsOid::compute(b"hello\n");
94 assert_eq!(
95 oid.as_str(),
96 "5891b5b522d5df086d0ff0b110fbd9d21bb4fc7163af34d08286a2e846f6be03"
97 );
98 }
99}
No newline at end of file
Added lib/gitlfs/store.rs lib/gitlfs/store.rs +377 −0 View lib/gitlfs/store.rs
@@ -0,0 +1,377 @@
1//! LFS object store: content-addressed, deduplicated persistence over `lib/storage`.
2
3use std::collections::HashSet;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use bytes::Bytes;
9use futures::stream::BoxStream;
10use futures::StreamExt;
11use object_store::path::Path;
12use object_store::PutPayload;
13use sha2::{Digest, Sha256};
14use storage::StorageRepository;
15use url::Url;
16
17use crate::error::LfsError;
18use crate::oid::LfsOid;
19
20/// Abstract LFS object access for protocol layers (`gitsmart`, `gitssh`).
21#[async_trait]
22pub trait LfsObjectStore: Send + Sync {
23 /// Store bytes under `oid`, verifying sha256(oid) == oid before writing.
24 async fn put_object(&self, oid: &LfsOid, data: Bytes) -> Result<(), LfsError>;
25 /// Fetch an object's raw bytes.
26 async fn get_object(&self, oid: &LfsOid) -> Result<Bytes, LfsError>;
27 /// Whether an object is present.
28 async fn has_object(&self, oid: &LfsOid) -> Result<bool, LfsError>;
29 /// Size of an object in bytes.
30 async fn object_size(&self, oid: &LfsOid) -> Result<u64, LfsError>;
31 /// Delete an object.
32 async fn delete_object(&self, oid: &LfsOid) -> Result<(), LfsError>;
33 /// List all stored oids.
34 async fn list_oids(&self) -> Result<Vec<LfsOid>, LfsError>;
35 /// Remove every object not in `keep`; returns how many were removed.
36 async fn cleanup(&self, keep: &HashSet<LfsOid>) -> Result<u64, LfsError>;
37 /// Presigned upload URL (unsupported by local backends).
38 async fn presign_upload(&self, oid: &LfsOid, expires_in: Duration) -> Result<Url, LfsError>;
39 /// Presigned download URL (unsupported by local backends).
40 async fn presign_download(&self, oid: &LfsOid, expires_in: Duration) -> Result<Url, LfsError>;
41 /// Stream an object's bytes into storage, verifying the sha256 as it flows.
42 async fn put_object_stream(
43 &self,
44 oid: &LfsOid,
45 stream: BoxStream<'static, Result<Bytes, LfsError>>,
46 ) -> Result<(), LfsError>;
47 /// Stream an object's bytes out of storage.
48 async fn get_object_stream(
49 &self,
50 oid: &LfsOid,
51 ) -> Result<BoxStream<'static, Result<Bytes, LfsError>>, LfsError>;
52}
53
54/// Default [`LfsObjectStore`] backed by an arbitrary [`StorageRepository`].
55///
56/// Objects are stored flat under `{prefix}/objects/{oid}`, so the same oid
57/// always maps to the same path regardless of repository — deduplication is
58/// inherent to the addressable layout.
59#[derive(Clone)]
60pub struct DefaultLfsStore {
61 storage: Arc<dyn StorageRepository>,
62 prefix: String,
63}
64
65impl DefaultLfsStore {
66 /// Create a store rooted at `prefix` (e.g. `"lfs"`).
67 pub fn new(storage: Arc<dyn StorageRepository>, prefix: impl Into<String>) -> Self {
68 Self {
69 storage,
70 prefix: prefix.into(),
71 }
72 }
73
74 fn object_path(&self, oid: &LfsOid) -> Path {
75 Path::from(format!("{}/objects/{}", self.prefix, oid.as_str()))
76 }
77}
78
79#[async_trait]
80impl LfsObjectStore for DefaultLfsStore {
81 async fn put_object(&self, oid: &LfsOid, data: Bytes) -> Result<(), LfsError> {
82 let actual = LfsOid::compute(&data);
83 if &actual != oid {
84 return Err(LfsError::HashMismatch {
85 expected: oid.as_str().to_string(),
86 actual: actual.as_str().to_string(),
87 });
88 }
89 self.storage.put(&self.object_path(oid), data).await?;
90 Ok(())
91 }
92
93 async fn get_object(&self, oid: &LfsOid) -> Result<Bytes, LfsError> {
94 let res = self.storage.get(&self.object_path(oid)).await?;
95 let bytes = res.bytes().await.map_err(storage::StorageError::from)?;
96 Ok(bytes)
97 }
98
99 async fn has_object(&self, oid: &LfsOid) -> Result<bool, LfsError> {
100 Ok(self.storage.exists(&self.object_path(oid)).await?)
101 }
102
103 async fn object_size(&self, oid: &LfsOid) -> Result<u64, LfsError> {
104 let meta = self.storage.head(&self.object_path(oid)).await?;
105 Ok(meta.size)
106 }
107
108 async fn delete_object(&self, oid: &LfsOid) -> Result<(), LfsError> {
109 self.storage.delete(&self.object_path(oid)).await?;
110 Ok(())
111 }
112
113 async fn list_oids(&self) -> Result<Vec<LfsOid>, LfsError> {
114 let prefix = Path::from(format!("{}/objects", self.prefix));
115 let metas = self.storage.list(Some(&prefix)).await?;
116 let mut oids = Vec::with_capacity(metas.len());
117 for meta in metas {
118 let name = meta.location.as_ref().rsplit('/').next().unwrap_or("");
119 if let Ok(oid) = LfsOid::new(name) {
120 oids.push(oid);
121 }
122 }
123 Ok(oids)
124 }
125
126 async fn cleanup(&self, keep: &HashSet<LfsOid>) -> Result<u64, LfsError> {
127 let mut removed = 0u64;
128 for oid in self.list_oids().await? {
129 if !keep.contains(&oid) {
130 self.storage.delete(&self.object_path(&oid)).await?;
131 removed += 1;
132 }
133 }
134 Ok(removed)
135 }
136
137 async fn presign_upload(&self, oid: &LfsOid, expires_in: Duration) -> Result<Url, LfsError> {
138 Ok(self
139 .storage
140 .presigned_upload(&self.object_path(oid), expires_in)
141 .await?)
142 }
143
144 async fn presign_download(&self, oid: &LfsOid, expires_in: Duration) -> Result<Url, LfsError> {
145 Ok(self
146 .storage
147 .presigned_download(&self.object_path(oid), expires_in)
148 .await?)
149 }
150
151 async fn put_object_stream(
152 &self,
153 oid: &LfsOid,
154 stream: BoxStream<'static, Result<Bytes, LfsError>>,
155 ) -> Result<(), LfsError> {
156 const PART_SIZE: usize = 5 * 1024 * 1024;
157
158 let path = self.object_path(oid);
159 let mut upload = self.storage.begin_multipart(&path).await?;
160 let mut hasher = Sha256::new();
161 let mut buf: Vec<u8> = Vec::new();
162 let mut sent_parts = 0usize;
163
164 futures::pin_mut!(stream);
165 while let Some(chunk) = stream.next().await {
166 let chunk = match chunk {
167 Ok(c) => c,
168 Err(e) => {
169 let _ = upload.abort().await;
170 return Err(e);
171 }
172 };
173 hasher.update(&chunk);
174 buf.extend_from_slice(&chunk);
175 if buf.len() >= PART_SIZE {
176 let part = std::mem::take(&mut buf);
177 if let Err(e) = upload.put_part(PutPayload::from_bytes(Bytes::from(part))).await {
178 let _ = upload.abort().await;
179 return Err(LfsError::Storage(e.into()));
180 }
181 sent_parts += 1;
182 }
183 }
184
185 if sent_parts == 0 && buf.is_empty() {
186 // Multipart cannot complete with zero parts; store empty object directly.
187 let _ = upload.abort().await;
188 self.storage
189 .put(&path, Bytes::new())
190 .await
191 .map_err(LfsError::Storage)?;
192 return Ok(());
193 }
194
195 if !buf.is_empty() {
196 if let Err(e) = upload.put_part(PutPayload::from_bytes(Bytes::from(buf))).await {
197 let _ = upload.abort().await;
198 return Err(LfsError::Storage(e.into()));
199 }
200 }
201
202 let actual = LfsOid::from_hasher(hasher);
203 if &actual != oid {
204 let _ = upload.abort().await;
205 return Err(LfsError::HashMismatch {
206 expected: oid.as_str().to_string(),
207 actual: actual.as_str().to_string(),
208 });
209 }
210
211 upload.complete().await.map_err(|e| LfsError::Storage(e.into()))?;
212 Ok(())
213 }
214
215 async fn get_object_stream(
216 &self,
217 oid: &LfsOid,
218 ) -> Result<BoxStream<'static, Result<Bytes, LfsError>>, LfsError> {
219 let res = self.storage.get(&self.object_path(oid)).await.map_err(|e| {
220 if matches!(&e, storage::StorageError::ObjectStore(object_store::Error::NotFound { .. })) {
221 LfsError::NotFound(oid.as_str().to_string())
222 } else {
223 LfsError::Storage(e)
224 }
225 })?;
226 Ok(res
227 .into_stream()
228 .map(|r| r.map_err(|e| LfsError::Storage(storage::StorageError::from(e))))
229 .boxed())
230 }
231}
232
233#[cfg(test)]
234mod tests {
235 use super::*;
236 use storage::config::{LocalStorageSettings, StorageSettings};
237
238 fn make_store() -> (tempfile::TempDir, DefaultLfsStore) {
239 let dir = tempfile::tempdir().expect("tempdir");
240 let mut settings = StorageSettings::default();
241 settings.set_local(
242 LocalStorageSettings::builder()
243 .root_dir(dir.path().join("store"))
244 .build(),
245 );
246 let storage = storage::repo::from_config(&settings).expect("storage");
247 (dir, DefaultLfsStore::new(storage, "lfs"))
248 }
249
250 #[tokio::test]
251 async fn put_get_roundtrip() {
252 let (_dir, store) = make_store();
253 let data = Bytes::from_static(b"hello lfs");
254 let oid = LfsOid::compute(&data);
255 store.put_object(&oid, data.clone()).await.unwrap();
256 assert!(store.has_object(&oid).await.unwrap());
257 assert_eq!(store.object_size(&oid).await.unwrap(), data.len() as u64);
258 assert_eq!(store.get_object(&oid).await.unwrap(), data);
259 assert_eq!(store.list_oids().await.unwrap(), vec![oid.clone()]);
260 store.delete_object(&oid).await.unwrap();
261 assert!(!store.has_object(&oid).await.unwrap());
262 }
263
264 #[tokio::test]
265 async fn put_rejects_hash_mismatch() {
266 let (_dir, store) = make_store();
267 let oid = LfsOid::new(&"0".repeat(64)).unwrap();
268 let err = store
269 .put_object(&oid, Bytes::from_static(b"wrong content"))
270 .await
271 .unwrap_err();
272 assert!(matches!(err, LfsError::HashMismatch { .. }));
273 assert!(!store.has_object(&oid).await.unwrap());
274 }
275
276 #[tokio::test]
277 async fn cleanup_removes_unkept_only() {
278 let (_dir, store) = make_store();
279 let a = Bytes::from_static(b"a");
280 let b = Bytes::from_static(b"b");
281 let oid_a = LfsOid::compute(&a);
282 let oid_b = LfsOid::compute(&b);
283 store.put_object(&oid_a, a).await.unwrap();
284 store.put_object(&oid_b, b).await.unwrap();
285
286 let mut keep = HashSet::new();
287 keep.insert(oid_a.clone());
288 assert_eq!(store.cleanup(&keep).await.unwrap(), 1);
289 assert!(store.has_object(&oid_a).await.unwrap());
290 assert!(!store.has_object(&oid_b).await.unwrap());
291 }
292
293 #[tokio::test]
294 async fn get_missing_returns_error() {
295 let (_dir, store) = make_store();
296 let oid = LfsOid::new(&"0".repeat(64)).unwrap();
297 assert!(store.get_object(&oid).await.is_err());
298 assert!(!store.has_object(&oid).await.unwrap());
299 }
300
301 #[tokio::test]
302 async fn presign_unsupported_on_local() {
303 let (_dir, store) = make_store();
304 let oid = LfsOid::new(&"0".repeat(64)).unwrap();
305 let err = store
306 .presign_upload(&oid, Duration::from_secs(60))
307 .await
308 .unwrap_err();
309 assert!(matches!(
310 err,
311 LfsError::Storage(storage::StorageError::Unsupported(_))
312 ));
313 }
314
315 #[tokio::test]
316 async fn put_object_stream_roundtrip() {
317 let (_dir, store) = make_store();
318 let data = Bytes::from_static(b"hello streaming lfs");
319 let oid = LfsOid::compute(&data);
320 let stream = futures::stream::iter(vec![
321 Ok(Bytes::from_static(b"hello ")),
322 Ok(Bytes::from_static(b"streaming lfs")),
323 ]);
324 store
325 .put_object_stream(&oid, stream.boxed())
326 .await
327 .unwrap();
328
329 let mut got = store.get_object_stream(&oid).await.unwrap();
330 let mut collected = Vec::new();
331 while let Some(chunk) = got.next().await {
332 collected.extend_from_slice(&chunk.unwrap());
333 }
334 assert_eq!(collected, data.to_vec());
335 }
336
337 #[tokio::test]
338 async fn put_object_stream_rejects_hash_mismatch() {
339 let (_dir, store) = make_store();
340 let oid = LfsOid::new(&"0".repeat(64)).unwrap();
341 let stream = futures::stream::iter(vec![Ok(Bytes::from_static(b"whatever"))]);
342 let err = store
343 .put_object_stream(&oid, stream.boxed())
344 .await
345 .unwrap_err();
346 assert!(matches!(err, LfsError::HashMismatch { .. }));
347 assert!(!store.has_object(&oid).await.unwrap());
348 }
349
350 #[tokio::test]
351 async fn put_object_stream_large() {
352 let (_dir, store) = make_store();
353 let data = vec![0xabu8; 6 * 1024 * 1024];
354 let oid = LfsOid::compute(&data);
355 let stream = futures::stream::iter(
356 data.chunks(1024 * 1024)
357 .map(|c| Ok(Bytes::copy_from_slice(c)))
358 .collect::<Vec<_>>(),
359 );
360 store.put_object_stream(&oid, stream.boxed()).await.unwrap();
361 assert_eq!(store.object_size(&oid).await.unwrap(), data.len() as u64);
362
363 let mut got = store.get_object_stream(&oid).await.unwrap();
364 let mut collected = Vec::new();
365 while let Some(chunk) = got.next().await {
366 collected.extend_from_slice(&chunk.unwrap());
367 }
368 assert_eq!(collected, data);
369 }
370
371 #[tokio::test]
372 async fn get_object_stream_missing() {
373 let (_dir, store) = make_store();
374 let oid = LfsOid::new(&"0".repeat(64)).unwrap();
375 assert!(store.get_object_stream(&oid).await.is_err());
376 }
377}
No newline at end of file