mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-29 16:37:07 +00:00
refactor(ecstore): drop the client shim, import rustfs-s3-client directly (#6668)
* refactor(ecstore): drop the client shim, import rustfs-s3-client directly Completes the migration window opened by the rustfs-s3-client extraction (rustfs/backlog#1842 PR3): every consumer now imports the client crate directly and the crate::client shim is deleted. - All in-crate crate::client:: paths (tier warm backends, tier core, lifecycle tier_sweeper, replication storage boundary, set_disk) now import rustfs_s3_client::* directly; crates/ecstore/src/client/mod.rs and the lib.rs mod client declaration are gone. - The two server-side modules historically misfiled under client/ move to their real homes: object_api_utils.rs to crates/ecstore/src/object_api/ (it builds engine-side object readers/writers), and object_handlers_common.rs to crates/ecstore/src/bucket/lifecycle/ (it is the lifecycle noncurrent-version cleanup helper). The latter now routes its replication calls through the lifecycle replication_sink boundary (schedule_delete wrapper and the sink's ReplicationObjectBridge re-export), as the lifecycle guard requires. - The ecstore public facade drops api::client: object_api_utils is exposed as api::object_api_utils, and the rustfs crate takes admin_handler_utils (AdminError) from rustfs-s3-client directly (new dependency). - Guard updates: the migration guard no longer pins mod client in ecstore's lib.rs or the admin_handler_utils facade module (it pins the new api::object_api_utils facade instead), and the module-lint register follows object_api_utils.rs to its new path. Verification: cargo check -p rustfs-ecstore --all-targets and -p rustfs; cargo fmt --all; tier/transition/lifecycle-focused nextest (626 passed) and the decommission/rebalance/heal families in a filtered run (603 passed; the full-suite parallel run only fails on this machine's known decommission/rebalance baseline flakes, which pass in filtered reruns and fail identically on pristine origin/main); layer/migration/s3s/logging/error-format/doc-path guard scripts all pass. * docs(architecture): record the S3 client extraction and reword invariant 4 (#6669) Closes the documentation step of rustfs/backlog#1842. ARCHITECTURE.md invariant 4 now states the serving-vs-consuming distinction the adversarial ruling asked for: ecstore must not serve HTTP/S3 wire types, while consuming remote S3 endpoints is a legitimate engine capability that lives in the extracted rustfs-s3-client crate. The violation note is updated from the pre-extraction snapshot (58 files, embedded client) to the current ratcheted state (shrink-only S3S_ECSTORE_FILES_BASELINE in scripts/check_s3s_footprint.sh, object_lock converted first), and the crate map gains s3-client. ecstore-module-split-plan.md gets the client-directory entry the plan was missing: a Current Shape row and a completed-extraction section describing the pure-move + shim + direct-import sequence and the re-homing of the two misfiled server-side modules.
This commit is contained in:
@@ -14,6 +14,8 @@
|
||||
|
||||
// #730: object API readers keep staged compatibility paths during facade migration.
|
||||
|
||||
pub mod object_api_utils;
|
||||
|
||||
use crate::bucket::metadata_sys::get_versioning_config;
|
||||
use crate::bucket::replication::{
|
||||
DeleteReplicationConfigSnapshot, ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,
|
||||
|
||||
@@ -0,0 +1,278 @@
|
||||
#![allow(clippy::map_entry)]
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
#![allow(unused_imports)]
|
||||
#![allow(unused_variables)]
|
||||
#![allow(unused_mut)]
|
||||
#![allow(unused_assignments)]
|
||||
#![allow(unused_must_use)]
|
||||
#![allow(clippy::all)]
|
||||
|
||||
use http::HeaderMap;
|
||||
use s3s::dto::ETag;
|
||||
use std::{collections::HashMap, io::Cursor, sync::Arc};
|
||||
use tokio::io::BufReader;
|
||||
|
||||
use crate::error::ErrorResponse;
|
||||
use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
|
||||
use crate::storage_api_contracts::range::HTTPRangeSpec;
|
||||
use rustfs_filemeta::ObjectPartInfo;
|
||||
use rustfs_rio::HashReader;
|
||||
use s3s::S3ErrorCode;
|
||||
|
||||
//#[derive(Clone)]
|
||||
pub struct PutObjReader {
|
||||
pub reader: HashReader,
|
||||
//pub sealMD5Fn: SealMD5CurrFn,
|
||||
}
|
||||
|
||||
impl PutObjReader {
|
||||
pub fn new(reader: HashReader) -> Self {
|
||||
Self { reader }
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn md5_current_hex_string(&self) -> String {
|
||||
self.reader.checksum().map(|v| v.encoded).unwrap_or_default()
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn with_encryption(&mut self, enc_reader: HashReader) -> Result<(), std::io::Error> {
|
||||
self.reader = enc_reader;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub type ObjReaderFn<'a> = Arc<dyn Fn(BufReader<Cursor<Vec<u8>>>, HeaderMap) -> GetObjectReader + Send + Sync + 'a>;
|
||||
|
||||
fn part_number_to_rangespec(oi: ObjectInfo, part_number: usize) -> Option<HTTPRangeSpec> {
|
||||
if oi.size == 0 || oi.parts.len() == 0 {
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut start: i64 = 0;
|
||||
let mut end: i64 = -1;
|
||||
let mut i = 0;
|
||||
while i < oi.parts.len() && i < part_number {
|
||||
start = end + 1;
|
||||
end = start + oi.parts[i].actual_size as i64 - 1;
|
||||
i += 1;
|
||||
}
|
||||
|
||||
Some(HTTPRangeSpec {
|
||||
start,
|
||||
end,
|
||||
is_suffix_length: false,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn new_getobjectreader<'a>(
|
||||
rs: &Option<HTTPRangeSpec>,
|
||||
oi: &'a ObjectInfo,
|
||||
opts: &ObjectOptions,
|
||||
_h: &HeaderMap,
|
||||
) -> Result<(ObjReaderFn<'a>, i64, i64), ErrorResponse> {
|
||||
//let (_, mut is_encrypted) = crypto.is_encrypted(oi.user_defined)?;
|
||||
let mut is_encrypted = false;
|
||||
let is_compressed = false; //oi.is_compressed_ok();
|
||||
|
||||
let rs_;
|
||||
if rs.is_none()
|
||||
&& let Some(part_number) = opts.part_number
|
||||
&& part_number > 0
|
||||
{
|
||||
rs_ = part_number_to_rangespec(oi.clone(), part_number);
|
||||
} else {
|
||||
rs_ = rs.clone();
|
||||
}
|
||||
|
||||
let mut get_fn: ObjReaderFn;
|
||||
|
||||
if let Some(rs_) = rs_ {
|
||||
let (off, length) = match rs_.get_offset_length(oi.size) {
|
||||
Ok(x) => x,
|
||||
Err(err) => {
|
||||
return Err(ErrorResponse {
|
||||
code: S3ErrorCode::InvalidRange,
|
||||
message: err.to_string(),
|
||||
key: None,
|
||||
bucket_name: None,
|
||||
region: None,
|
||||
request_id: None,
|
||||
host_id: "".to_string(),
|
||||
});
|
||||
}
|
||||
};
|
||||
get_fn = Arc::new(move |input_reader: BufReader<Cursor<Vec<u8>>>, _: HeaderMap| {
|
||||
//Box::pin({
|
||||
let r = GetObjectReader {
|
||||
object_info: oi.clone(),
|
||||
stream: Box::new(input_reader),
|
||||
buffered_body: None,
|
||||
body_source: Default::default(),
|
||||
};
|
||||
r
|
||||
//})
|
||||
});
|
||||
|
||||
return Ok((get_fn, off as i64, length as i64));
|
||||
}
|
||||
if rs.is_none() && opts.part_number.is_none() && oi.size >= 0 {
|
||||
get_fn = Arc::new(move |input_reader: BufReader<Cursor<Vec<u8>>>, _: HeaderMap| GetObjectReader {
|
||||
object_info: oi.clone(),
|
||||
stream: Box::new(input_reader),
|
||||
buffered_body: None,
|
||||
body_source: Default::default(),
|
||||
});
|
||||
|
||||
return Ok((get_fn, 0, oi.size));
|
||||
}
|
||||
Err(ErrorResponse {
|
||||
code: S3ErrorCode::InvalidRange,
|
||||
message: "Invalid range".to_string(),
|
||||
key: Some(oi.name.clone()),
|
||||
bucket_name: Some(oi.bucket.clone()),
|
||||
region: Some("".to_string()),
|
||||
request_id: None,
|
||||
host_id: "".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Convert a raw stored ETag into the strongly-typed `s3s::dto::ETag`.
|
||||
///
|
||||
/// Supports already quoted (`"abc"`), weak (`W/"abc"`), or plain (`abc`) values.
|
||||
pub fn to_s3s_etag(etag: &str) -> ETag {
|
||||
if let Some(rest) = etag.strip_prefix("W/\"") {
|
||||
if let Some(body) = rest.strip_suffix('"') {
|
||||
return ETag::Weak(body.to_string());
|
||||
}
|
||||
return ETag::Weak(rest.to_string());
|
||||
}
|
||||
|
||||
if let Some(body) = etag.strip_prefix('"').and_then(|rest| rest.strip_suffix('"')) {
|
||||
return ETag::Strong(body.to_string());
|
||||
}
|
||||
|
||||
ETag::Strong(etag.to_string())
|
||||
}
|
||||
|
||||
pub fn get_raw_etag(metadata: &HashMap<String, String>) -> String {
|
||||
metadata
|
||||
.get("etag")
|
||||
.cloned()
|
||||
.or_else(|| metadata.get("md5Sum").cloned())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::Arc;
|
||||
|
||||
fn multipart_object_info() -> ObjectInfo {
|
||||
ObjectInfo {
|
||||
bucket: "bucket".to_string(),
|
||||
name: "object".to_string(),
|
||||
size: 6 * 1024 * 1024,
|
||||
actual_size: 6 * 1024 * 1024,
|
||||
parts: Arc::new(vec![
|
||||
ObjectPartInfo {
|
||||
number: 1,
|
||||
size: 5 * 1024 * 1024,
|
||||
actual_size: 5 * 1024 * 1024,
|
||||
..Default::default()
|
||||
},
|
||||
ObjectPartInfo {
|
||||
number: 2,
|
||||
size: 1024 * 1024,
|
||||
actual_size: 1024 * 1024,
|
||||
..Default::default()
|
||||
},
|
||||
]),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_new_getobjectreader_uses_full_range_for_unranged_multipart_object() {
|
||||
let oi = multipart_object_info();
|
||||
let opts = ObjectOptions::default();
|
||||
|
||||
let result = new_getobjectreader(&None, &oi, &opts, &HeaderMap::new())
|
||||
.expect("unranged multipart object should build a full-object reader");
|
||||
|
||||
assert_eq!(result.1, 0);
|
||||
assert_eq!(result.2, oi.size);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_new_getobjectreader_keeps_part_number_range_for_multipart_object() {
|
||||
let oi = multipart_object_info();
|
||||
let opts = ObjectOptions {
|
||||
part_number: Some(2),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let result = new_getobjectreader(&None, &oi, &opts, &HeaderMap::new()).expect("part number should build that part range");
|
||||
|
||||
assert_eq!(result.1, 5 * 1024 * 1024);
|
||||
assert_eq!(result.2, 1024 * 1024);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_to_s3s_etag() {
|
||||
// Test unquoted ETag - should become strong etag
|
||||
assert_eq!(
|
||||
to_s3s_etag("6af8d12c0c74b78094884349f3c8a079"),
|
||||
ETag::Strong("6af8d12c0c74b78094884349f3c8a079".to_string())
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
to_s3s_etag("\"6af8d12c0c74b78094884349f3c8a079\""),
|
||||
ETag::Strong("6af8d12c0c74b78094884349f3c8a079".to_string())
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
to_s3s_etag("W/\"6af8d12c0c74b78094884349f3c8a079\""),
|
||||
ETag::Weak("6af8d12c0c74b78094884349f3c8a079".to_string())
|
||||
);
|
||||
|
||||
assert_eq!(to_s3s_etag(""), ETag::Strong(String::new()));
|
||||
|
||||
assert_eq!(to_s3s_etag("\"incomplete"), ETag::Strong("\"incomplete".to_string()));
|
||||
|
||||
assert_eq!(to_s3s_etag("incomplete\""), ETag::Strong("incomplete\"".to_string()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_extract_etag() {
|
||||
let mut metadata = HashMap::new();
|
||||
|
||||
// Test with etag field
|
||||
metadata.insert("etag".to_string(), "abc123".to_string());
|
||||
assert_eq!(get_raw_etag(&metadata), "abc123");
|
||||
|
||||
metadata.insert("etag".to_string(), "\"def456\"".to_string());
|
||||
assert_eq!(get_raw_etag(&metadata), "\"def456\"");
|
||||
|
||||
// Test fallback to md5Sum
|
||||
metadata.remove("etag");
|
||||
metadata.insert("md5Sum".to_string(), "xyz789".to_string());
|
||||
assert_eq!(get_raw_etag(&metadata), "xyz789");
|
||||
|
||||
metadata.clear();
|
||||
assert_eq!(get_raw_etag(&metadata), "");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user