Skip to content

Commit 6aaf5cf

Browse files
committed
Add xattr-based attribute support to LocalFileSystem
Store object attributes as extended attributes on the local filesystem, enabling put_opts and put_multipart_opts to persist attributes. Attributes are set on the staging file before atomic rename to preserve consistency. - Add xattr dependency (gated by fs feature) - Map standard attributes to user.* xattr namespace - Read attributes back in get_opts
1 parent 8ef1aaa commit 6aaf5cf

2 files changed

Lines changed: 204 additions & 22 deletions

File tree

Cargo.toml

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,9 @@ serde_json = { version = "1.0", default-features = false, features = ["std"], op
6262
serde_urlencoded = { version = "0.7", optional = true }
6363
tokio = { version = "1.29.0", features = ["sync", "macros", "rt", "time", "io-util"] }
6464

65+
[target.'cfg(target_family="unix")'.dependencies]
66+
xattr = { version = "1", optional = true }
67+
6568
[target.'cfg(target_family="unix")'.dev-dependencies]
6669
nix = { version = "0.30.0", features = ["fs"] }
6770

@@ -73,7 +76,7 @@ wasm-bindgen-futures = "0.4.18"
7376
default = ["fs"]
7477
cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/stream", "chrono/serde", "base64", "rand", "ring", "http-body-util", "form_urlencoded", "serde_urlencoded"]
7578
azure = ["cloud", "httparse"]
76-
fs = ["walkdir"]
79+
fs = ["walkdir", "xattr"]
7780
gcp = ["cloud", "rustls-pki-types"]
7881
aws = ["cloud", "md-5"]
7982
http = ["cloud"]

src/local.rs

Lines changed: 200 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
// under the License.
1717

1818
//! An object store implementation for a local filesystem
19+
#[cfg(target_family = "unix")]
20+
use std::borrow::Cow;
1921
use std::fs::{File, Metadata, OpenOptions, metadata, symlink_metadata};
2022
use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
2123
use std::ops::Range;
@@ -34,9 +36,9 @@ use url::Url;
3436
use walkdir::{DirEntry, WalkDir};
3537

3638
use crate::{
37-
Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
38-
ObjectStore, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
39-
UploadPart, maybe_spawn_blocking,
39+
Attribute, Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload,
40+
ObjectMeta, ObjectStore, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult,
41+
Result, UploadPart, maybe_spawn_blocking,
4042
path::{Path, absolute_path_to_url},
4143
util::InvalidGetRange,
4244
};
@@ -112,6 +114,20 @@ pub(crate) enum Error {
112114

113115
#[error("Upload aborted")]
114116
Aborted,
117+
118+
#[cfg(target_family = "unix")]
119+
#[error("Unable to set extended attribute on {}: {source}", path.display())]
120+
UnableToSetXattr {
121+
source: std::io::Error,
122+
path: PathBuf,
123+
},
124+
125+
#[cfg(target_family = "unix")]
126+
#[error("Unable to read extended attribute on {}: {source}", path.display())]
127+
UnableToReadXattr {
128+
source: std::io::Error,
129+
path: PathBuf,
130+
},
115131
}
116132

117133
impl From<Error> for super::Error {
@@ -321,6 +337,98 @@ fn is_valid_file_path(path: &Path) -> bool {
321337
}
322338
}
323339

340+
/// Sets extended attributes on a file from an Attributes collection
341+
#[cfg(target_family = "unix")]
342+
fn set_xattrs(path: &std::path::Path, attributes: &Attributes) -> Result<()> {
343+
for (attr, value) in attributes {
344+
let name: Cow<'static, str> = match attr {
345+
Attribute::CacheControl => Cow::Borrowed("user.cache_control"),
346+
Attribute::ContentDisposition => Cow::Borrowed("user.content_disposition"),
347+
Attribute::ContentEncoding => Cow::Borrowed("user.content_encoding"),
348+
Attribute::ContentLanguage => Cow::Borrowed("user.content_language"),
349+
Attribute::ContentType => Cow::Borrowed("user.content_type"),
350+
Attribute::StorageClass => Cow::Borrowed("user.storage_class"),
351+
Attribute::Metadata(key) => Cow::Owned(format!("user.{key}")),
352+
};
353+
xattr::set(path, name.as_ref(), value.as_ref().as_bytes()).map_err(|source| {
354+
Error::UnableToSetXattr {
355+
source,
356+
path: path.into(),
357+
}
358+
})?;
359+
}
360+
Ok(())
361+
}
362+
363+
/// Reads extended attributes from a file and returns an Attributes collection
364+
#[cfg(target_family = "unix")]
365+
fn get_xattrs(path: &std::path::Path) -> Result<Attributes> {
366+
let mut attributes = Attributes::new();
367+
368+
let list = match xattr::list(path) {
369+
Ok(list) => list,
370+
Err(source) => {
371+
return Err(Error::UnableToReadXattr {
372+
source,
373+
path: path.into(),
374+
}
375+
.into());
376+
}
377+
};
378+
379+
for name in list {
380+
let name_str = match name.to_str() {
381+
Some(s) if s.starts_with("user.") => s,
382+
_ => continue,
383+
};
384+
385+
let value = match xattr::get(path, &name) {
386+
Ok(Some(v)) => match String::from_utf8(v) {
387+
Ok(s) => s,
388+
Err(_) => continue,
389+
},
390+
Ok(None) => continue,
391+
Err(source) => {
392+
return Err(Error::UnableToReadXattr {
393+
source,
394+
path: path.into(),
395+
}
396+
.into());
397+
}
398+
};
399+
400+
let key = &name_str["user.".len()..];
401+
let attr = match key {
402+
"cache_control" => Attribute::CacheControl,
403+
"content_disposition" => Attribute::ContentDisposition,
404+
"content_encoding" => Attribute::ContentEncoding,
405+
"content_language" => Attribute::ContentLanguage,
406+
"content_type" => Attribute::ContentType,
407+
"storage_class" => Attribute::StorageClass,
408+
key => Attribute::Metadata(key.to_string().into()),
409+
};
410+
attributes.insert(attr, value.into());
411+
}
412+
Ok(attributes)
413+
}
414+
415+
/// Returns an error if attributes are non-empty on non-Unix platforms.
416+
#[cfg(not(target_family = "unix"))]
417+
fn set_xattrs(_path: &std::path::Path, attributes: &Attributes) -> Result<()> {
418+
if !attributes.is_empty() {
419+
return Err(super::Error::NotSupported {
420+
source: "Setting extended attributes is only supported on Unix platforms".into(),
421+
});
422+
}
423+
Ok(())
424+
}
425+
426+
/// No-op on non-Unix platforms: returns empty attributes.
427+
#[cfg(not(target_family = "unix"))]
428+
fn get_xattrs(_path: &std::path::Path) -> Result<Attributes> {
429+
Ok(Attributes::new())
430+
}
431+
324432
#[async_trait]
325433
impl ObjectStore for LocalFileSystem {
326434
async fn put_opts(
@@ -336,13 +444,6 @@ impl ObjectStore for LocalFileSystem {
336444
});
337445
}
338446

339-
if !opts.attributes.is_empty() {
340-
return Err(crate::Error::NotImplemented {
341-
operation: "`put_opts` with `opts.attributes` specified".into(),
342-
implementer: self.to_string(),
343-
});
344-
}
345-
346447
let path = self.path_to_filesystem(location)?;
347448
maybe_spawn_blocking(move || {
348449
let (mut file, staging_path) = new_staged_upload(&path)?;
@@ -355,6 +456,11 @@ impl ObjectStore for LocalFileSystem {
355456
path: path.to_string_lossy().to_string(),
356457
})?;
357458
e_tag = Some(get_etag(&metadata));
459+
460+
if !opts.attributes.is_empty() {
461+
set_xattrs(&staging_path, &opts.attributes)?;
462+
}
463+
358464
match opts.mode {
359465
PutMode::Overwrite => {
360466
// For some fuse types of file systems, the file must be closed first
@@ -402,16 +508,9 @@ impl ObjectStore for LocalFileSystem {
402508
location: &Path,
403509
opts: PutMultipartOptions,
404510
) -> Result<Box<dyn MultipartUpload>> {
405-
if !opts.attributes.is_empty() {
406-
return Err(crate::Error::NotImplemented {
407-
operation: "`put_multipart_opts` with `opts.attributes` specified".into(),
408-
implementer: self.to_string(),
409-
});
410-
}
411-
412511
let dest = self.path_to_filesystem(location)?;
413512
let (file, src) = new_staged_upload(&dest)?;
414-
Ok(Box::new(LocalUpload::new(src, dest, file)))
513+
Ok(Box::new(LocalUpload::new(src, dest, file, opts.attributes)))
415514
}
416515

417516
async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
@@ -429,9 +528,11 @@ impl ObjectStore for LocalFileSystem {
429528
None => 0..meta.size,
430529
};
431530

531+
let attributes = get_xattrs(&path)?;
532+
432533
Ok(GetResult {
433534
payload: GetResultPayload::File(file, path),
434-
attributes: Attributes::default(),
535+
attributes,
435536
range,
436537
meta,
437538
})
@@ -829,6 +930,8 @@ struct LocalUpload {
829930
src: Option<PathBuf>,
830931
/// The next offset to write into the file
831932
offset: u64,
933+
/// Attributes to set on the file
934+
attributes: Attributes,
832935
}
833936

834937
#[derive(Debug)]
@@ -838,14 +941,15 @@ struct UploadState {
838941
}
839942

840943
impl LocalUpload {
841-
pub(crate) fn new(src: PathBuf, dest: PathBuf, file: File) -> Self {
944+
pub(crate) fn new(src: PathBuf, dest: PathBuf, file: File, attributes: Attributes) -> Self {
842945
Self {
843946
state: Arc::new(UploadState {
844947
dest,
845948
file: Mutex::new(file),
846949
}),
847950
src: Some(src),
848951
offset: 0,
952+
attributes,
849953
}
850954
}
851955
}
@@ -876,9 +980,15 @@ impl MultipartUpload for LocalUpload {
876980
async fn complete(&mut self) -> Result<PutResult> {
877981
let src = self.src.take().ok_or(Error::Aborted)?;
878982
let s = Arc::clone(&self.state);
983+
let attributes = std::mem::take(&mut self.attributes);
879984
maybe_spawn_blocking(move || {
880985
// Ensure no inflight writes
881986
let file = s.file.lock();
987+
988+
if !attributes.is_empty() {
989+
set_xattrs(&src, &attributes)?;
990+
}
991+
882992
std::fs::rename(&src, &s.dest)
883993
.map_err(|source| Error::UnableToRenameFile { source })?;
884994
let metadata = file.metadata().map_err(|e| Error::Metadata {
@@ -1177,6 +1287,7 @@ mod tests {
11771287
copy_rename_nonexistent_object(&integration).await;
11781288
stream_get(&integration).await;
11791289
put_opts(&integration, false).await;
1290+
put_get_attributes(&integration).await;
11801291
}
11811292

11821293
#[test]
@@ -1790,7 +1901,7 @@ mod unix_test {
17901901
use tempfile::TempDir;
17911902

17921903
use crate::local::LocalFileSystem;
1793-
use crate::{ObjectStoreExt, Path};
1904+
use crate::{Attribute, Attributes, ObjectStore, ObjectStoreExt, Path, PutOptions};
17941905

17951906
#[tokio::test]
17961907
async fn test_fifo() {
@@ -1810,4 +1921,72 @@ mod unix_test {
18101921

18111922
spawned.await.unwrap();
18121923
}
1924+
1925+
#[tokio::test]
1926+
async fn test_put_get_attributes() {
1927+
let root = TempDir::new().unwrap();
1928+
let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
1929+
1930+
let location = Path::from("test_file");
1931+
let data = "test data";
1932+
1933+
let mut attributes = Attributes::new();
1934+
attributes.insert(Attribute::ContentType, "text/plain".into());
1935+
attributes.insert(Attribute::CacheControl, "max-age=3600".into());
1936+
attributes.insert(Attribute::ContentDisposition, "inline".into());
1937+
attributes.insert(Attribute::ContentEncoding, "gzip".into());
1938+
attributes.insert(Attribute::ContentLanguage, "en-US".into());
1939+
attributes.insert(Attribute::StorageClass, "STANDARD".into());
1940+
attributes.insert(
1941+
Attribute::Metadata("custom_key".into()),
1942+
"custom_value".into(),
1943+
);
1944+
1945+
let opts = PutOptions {
1946+
attributes: attributes.clone(),
1947+
..Default::default()
1948+
};
1949+
1950+
integration
1951+
.put_opts(&location, data.into(), opts)
1952+
.await
1953+
.unwrap();
1954+
1955+
let result = integration.get(&location).await.unwrap();
1956+
assert_eq!(result.attributes, attributes);
1957+
1958+
let bytes = result.bytes().await.unwrap();
1959+
assert_eq!(bytes.as_ref(), data.as_bytes());
1960+
}
1961+
1962+
#[tokio::test]
1963+
async fn test_multipart_attributes() {
1964+
let root = TempDir::new().unwrap();
1965+
let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
1966+
1967+
let location = Path::from("multipart_file");
1968+
1969+
let mut attributes = Attributes::new();
1970+
attributes.insert(Attribute::ContentType, "application/octet-stream".into());
1971+
attributes.insert(Attribute::Metadata("part_count".into()), "2".into());
1972+
1973+
let opts = crate::PutMultipartOptions {
1974+
attributes: attributes.clone(),
1975+
..Default::default()
1976+
};
1977+
1978+
let mut upload = integration
1979+
.put_multipart_opts(&location, opts)
1980+
.await
1981+
.unwrap();
1982+
upload.put_part("part1".into()).await.unwrap();
1983+
upload.put_part("part2".into()).await.unwrap();
1984+
upload.complete().await.unwrap();
1985+
1986+
let result = integration.get(&location).await.unwrap();
1987+
assert_eq!(result.attributes, attributes);
1988+
1989+
let bytes = result.bytes().await.unwrap();
1990+
assert_eq!(bytes.as_ref(), b"part1part2");
1991+
}
18131992
}

0 commit comments

Comments
 (0)