Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

refactor(gitlfs): refresh crate across oid, store, error, metrics

zhenyi committed
parent 90b42d8 commit a724114 Browse files

5 files changed

+213 −145

File tree 5
Modified lib/gitlfs/error.rs View lib/gitlfs/error.rs
@@ -1,6 +1,3 @@
1//! LFS storage-layer error type.
2
3/// Unified error for the `gitlfs` crate.
41#[derive(Debug, thiserror::Error)]
52pub enum LfsError {
63 #[error("invalid oid: {0}")]
Modified lib/gitlfs/lib.rs View lib/gitlfs/lib.rs
@@ -1,9 +1,5 @@
1//! LFS large-file storage core abstraction.
2//!
3//! Provides content-addressed (SHA-256 oid) object persistence over
4//! `lib/storage`, decoupled from any Git protocol layer.
5
61pub mod error;
2pub mod metrics;
73pub mod oid;
84pub mod store;
95
Added lib/gitlfs/metrics.rs View lib/gitlfs/metrics.rs
@@ -0,0 +1,45 @@
1use std::sync::Once;
2
3use metrics::{counter, describe_counter};
4
5use crate::error::LfsError;
6
7pub const OUTCOME_SUCCESS: &str = "success";
8pub const OP_PUT: &str = "put";
9pub const OP_GET: &str = "get";
10
11static DESCRIBE_ONCE: Once = Once::new();
12
13pub fn ensure_described() {
14 DESCRIBE_ONCE.call_once(describe_all);
15}
16
17fn describe_all() {
18 describe_counter!(
19 "gitlfs_store_ops_total",
20 "LFS store operations partitioned by op and outcome (success, error)."
21 );
22 describe_counter!(
23 "gitlfs_bytes_total",
24 "LFS object bytes transferred by op (put, get)."
25 );
26}
27
28fn outcome_for<T>(result: &Result<T, LfsError>) -> &'static str {
29 if result.is_ok() {
30 OUTCOME_SUCCESS
31 } else {
32 "error"
33 }
34}
35
36pub(crate) fn record_op<T>(op: &'static str, result: &Result<T, LfsError>) {
37 ensure_described();
38 counter!("gitlfs_store_ops_total", "op" => op, "outcome" => outcome_for(result))
39 .increment(1);
40}
41
42pub(crate) fn record_bytes(op: &'static str, n: usize) {
43 ensure_described();
44 counter!("gitlfs_bytes_total", "op" => op).increment(n as u64);
45}
No newline at end of file
Modified lib/gitlfs/oid.rs View lib/gitlfs/oid.rs
@@ -1,5 +1,3 @@
1//! LFS object identifier: a validated SHA-256 hex digest.
2
31use sha2::{Digest, Sha256};
42
53use crate::error::LfsError;
@@ -16,13 +14,11 @@
1614 out
1715}
1816
19/// A validated LFS object id (SHA-256, 64 lowercase hex chars).
2017#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2118pub struct LfsOid(String);
2219
2320impl LfsOid {
24 /// Validate a hex string and build an [`LfsOid`].
25 pub fn new(hex: impl AsRef<str>) -> Result<Self, LfsError> {
21 pub fn new(hex: impl AsRef<str>) -> Result<Self, LfsError> {
2622 let hex = hex.as_ref();
2723 if hex.len() != HEX_CHARS || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
2824 return Err(LfsError::InvalidOid(hex.to_string()));
@@ -30,18 +26,15 @@
3026 Ok(Self(hex.to_ascii_lowercase()))
3127 }
3228
33 /// Compute the SHA-256 oid of `data`.
34 pub fn compute(data: &[u8]) -> Self {
29 pub fn compute(data: &[u8]) -> Self {
3530 Self(hex_encode(&Sha256::digest(data)))
3631 }
3732
38 /// Build an [`LfsOid`] from an in-progress SHA-256 hasher.
39 pub fn from_hasher(hasher: Sha256) -> Self {
33 pub fn from_hasher(hasher: Sha256) -> Self {
4034 Self(hex_encode(&hasher.finalize()))
4135 }
4236
43 /// The 64-char lowercase hex representation.
44 pub fn as_str(&self) -> &str {
37 pub fn as_str(&self) -> &str {
4538 &self.0
4639 }
4740}
@@ -89,7 +82,6 @@
8982
9083 #[test]
9184 fn oid_compute_matches_sha256() {
92 // sha256(b"hello\n") = 5891b5b5...
9385 let oid = LfsOid::compute(b"hello\n");
9486 assert_eq!(
9587 oid.as_str(),
Modified lib/gitlfs/store.rs View lib/gitlfs/store.rs
@@ -1,5 +1,3 @@
1//! LFS object store: content-addressed, deduplicated persistence over `lib/storage`.
2
31use std::collections::HashSet;
42use std::sync::Arc;
53use std::time::Duration;
@@ -15,47 +13,31 @@
1513use url::Url;
1614
1715use crate::error::LfsError;
16use crate::metrics;
1817use crate::oid::LfsOid;
1918
20/// Abstract LFS object access for protocol layers (`gitsmart`, `gitssh`).
2119#[async_trait]
2220pub 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(
21 async fn put_object(&self, oid: &LfsOid, data: Bytes) -> Result<(), LfsError>;
22 async fn get_object(&self, oid: &LfsOid) -> Result<Bytes, LfsError>;
23 async fn has_object(&self, oid: &LfsOid) -> Result<bool, LfsError>;
24 async fn object_size(&self, oid: &LfsOid) -> Result<u64, LfsError>;
25 async fn delete_object(&self, oid: &LfsOid) -> Result<(), LfsError>;
26 async fn list_oids(&self) -> Result<Vec<LfsOid>, LfsError>;
27 async fn cleanup(&self, keep: &HashSet<LfsOid>) -> Result<u64, LfsError>;
28 async fn presign_upload(&self, oid: &LfsOid, expires_in: Duration) -> Result<Url, LfsError>;
29 async fn presign_download(&self, oid: &LfsOid, expires_in: Duration) -> Result<Url, LfsError>;
30 async fn put_object_stream(
4331 &self,
4432 oid: &LfsOid,
4533 stream: BoxStream<'static, Result<Bytes, LfsError>>,
4634 ) -> Result<(), LfsError>;
47 /// Stream an object's bytes out of storage.
48 async fn get_object_stream(
35 async fn get_object_stream(
4936 &self,
5037 oid: &LfsOid,
5138 ) -> Result<BoxStream<'static, Result<Bytes, LfsError>>, LfsError>;
5239}
5340
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.
5941#[derive(Clone)]
6042pub struct DefaultLfsStore {
6143 storage: Arc<dyn StorageRepository>,
@@ -63,8 +45,7 @@
6345}
6446
6547impl DefaultLfsStore {
66 /// Create a store rooted at `prefix` (e.g. `"lfs"`).
67 pub fn new(storage: Arc<dyn StorageRepository>, prefix: impl Into<String>) -> Self {
48 pub fn new(storage: Arc<dyn StorageRepository>, prefix: impl Into<String>) -> Self {
6849 Self {
6950 storage,
7051 prefix: prefix.into(),
@@ -79,73 +60,119 @@
7960#[async_trait]
8061impl LfsObjectStore for DefaultLfsStore {
8162 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 });
63 let len = data.len();
64 let result: Result<(), LfsError> = async {
65 let actual = LfsOid::compute(&data);
66 if &actual != oid {
67 return Err(LfsError::HashMismatch {
68 expected: oid.as_str().to_string(),
69 actual: actual.as_str().to_string(),
70 });
71 }
72 self.storage.put(&self.object_path(oid), data).await?;
73 Ok(())
8874 }
89 self.storage.put(&self.object_path(oid), data).await?;
90 Ok(())
75 .await;
76 metrics::record_bytes(metrics::OP_PUT, len);
77 metrics::record_op("put_object", &result);
78 result
9179 }
9280
9381 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)
82 let result: Result<Bytes, LfsError> = async {
83 let res = self.storage.get(&self.object_path(oid)).await?;
84 Ok(res.bytes().await.map_err(storage::StorageError::from)?)
85 }
86 .await;
87 metrics::record_bytes(metrics::OP_GET, result.as_ref().map_or(0, |b| b.len()));
88 metrics::record_op("get_object", &result);
89 result
9790 }
9891
9992 async fn has_object(&self, oid: &LfsOid) -> Result<bool, LfsError> {
100 Ok(self.storage.exists(&self.object_path(oid)).await?)
93 let result: Result<bool, LfsError> =
94 async { Ok(self.storage.exists(&self.object_path(oid)).await?) }
95 .await;
96 metrics::record_op("has_object", &result);
97 result
10198 }
10299
103100 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)
101 let result: Result<u64, LfsError> = async {
102 let meta = self.storage.head(&self.object_path(oid)).await?;
103 Ok(meta.size)
104 }
105 .await;
106 metrics::record_op("object_size", &result);
107 result
106108 }
107109
108110 async fn delete_object(&self, oid: &LfsOid) -> Result<(), LfsError> {
109 self.storage.delete(&self.object_path(oid)).await?;
110 Ok(())
111 let result: Result<(), LfsError> = async {
112 self.storage.delete(&self.object_path(oid)).await?;
113 Ok(())
114 }
115 .await;
116 metrics::record_op("delete_object", &result);
117 result
111118 }
112119
113120 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 let result: Result<Vec<LfsOid>, LfsError> = async {
122 let prefix = Path::from(format!("{}/objects", self.prefix));
123 let metas = self.storage.list(Some(&prefix)).await?;
124 let mut oids = Vec::with_capacity(metas.len());
125 for meta in metas {
126 let name = meta.location.as_ref().rsplit('/').next().unwrap_or("");
127 if let Ok(oid) = LfsOid::new(name) {
128 oids.push(oid);
129 }
121130 }
131 Ok(oids)
122132 }
123 Ok(oids)
133 .await;
134 metrics::record_op("list_oids", &result);
135 result
124136 }
125137
126138 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;
139 let result: Result<u64, LfsError> = async {
140 let mut removed = 0u64;
141 for oid in self.list_oids().await? {
142 if !keep.contains(&oid) {
143 self.storage.delete(&self.object_path(&oid)).await?;
144 removed += 1;
145 }
132146 }
147 Ok(removed)
133148 }
134 Ok(removed)
149 .await;
150 metrics::record_op("cleanup", &result);
151 result
135152 }
136153
137154 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?)
155 let result: Result<Url, LfsError> = async {
156 Ok(self
157 .storage
158 .presigned_upload(&self.object_path(oid), expires_in)
159 .await?)
160 }
161 .await;
162 metrics::record_op("presign_upload", &result);
163 result
142164 }
143165
144166 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?)
167 let result: Result<Url, LfsError> = async {
168 Ok(self
169 .storage
170 .presigned_download(&self.object_path(oid), expires_in)
171 .await?)
172 }
173 .await;
174 metrics::record_op("presign_download", &result);
175 result
149176 }
150177
151178 async fn put_object_stream(
@@ -155,78 +182,89 @@
155182 ) -> Result<(), LfsError> {
156183 const PART_SIZE: usize = 5 * 1024 * 1024;
157184
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;
185 let result: Result<(), LfsError> = async {
186 let path = self.object_path(oid);
187 let mut upload = self.storage.begin_multipart(&path).await?;
188 let mut hasher = Sha256::new();
189 let mut buf: Vec<u8> = Vec::new();
190 let mut sent_parts = 0usize;
163191
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);
192 futures::pin_mut!(stream);
193 while let Some(chunk) = stream.next().await {
194 let chunk = match chunk {
195 Ok(c) => c,
196 Err(e) => {
197 let _ = upload.abort().await;
198 return Err(e);
199 }
200 };
201 hasher.update(&chunk);
202 buf.extend_from_slice(&chunk);
203 if buf.len() >= PART_SIZE {
204 let part = std::mem::take(&mut buf);
205 if let Err(e) = upload.put_part(PutPayload::from_bytes(Bytes::from(part))).await
206 {
207 let _ = upload.abort().await;
208 return Err(LfsError::Storage(e.into()));
209 }
210 sent_parts += 1;
171211 }
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 {
212 }
213
214 if sent_parts == 0 && buf.is_empty() {
215 let _ = upload.abort().await;
216 self.storage
217 .put(&path, Bytes::new())
218 .await
219 .map_err(LfsError::Storage)?;
220 return Ok(());
221 }
222
223 if !buf.is_empty() {
224 if let Err(e) = upload.put_part(PutPayload::from_bytes(Bytes::from(buf))).await {
178225 let _ = upload.abort().await;
179226 return Err(LfsError::Storage(e.into()));
180227 }
181 sent_parts += 1;
182228 }
183 }
184229
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 {
230 let actual = LfsOid::from_hasher(hasher);
231 if &actual != oid {
197232 let _ = upload.abort().await;
198 return Err(LfsError::Storage(e.into()));
233 return Err(LfsError::HashMismatch {
234 expected: oid.as_str().to_string(),
235 actual: actual.as_str().to_string(),
236 });
199237 }
200 }
201238
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 });
239 upload.complete().await.map_err(|e| LfsError::Storage(e.into()))?;
240 Ok(())
209241 }
210
211 upload.complete().await.map_err(|e| LfsError::Storage(e.into()))?;
212 Ok(())
242 .await;
243 metrics::record_op("put_object_stream", &result);
244 result
213245 }
214246
215247 async fn get_object_stream(
216248 &self,
217249 oid: &LfsOid,
218250 ) -> 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())
251 let result: Result<BoxStream<'static, Result<Bytes, LfsError>>, LfsError> = async {
252 let res = self.storage.get(&self.object_path(oid)).await.map_err(|e| {
253 if matches!(&e, storage::StorageError::ObjectStore(object_store::Error::NotFound { .. }))
254 {
255 LfsError::NotFound(oid.as_str().to_string())
256 } else {
257 LfsError::Storage(e)
258 }
259 })?;
260 Ok(res
261 .into_stream()
262 .map(|r| r.map_err(|e| LfsError::Storage(storage::StorageError::from(e))))
263 .boxed())
264 }
265 .await;
266 metrics::record_op("get_object_stream", &result);
267 result
230268 }
231269}
232270