gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-25 13:36:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: converge stale TODOs and apply safe fills (backlog#646) (#4322)

Browse Source 
Second TODO-convergence round over the current tree (backlog#646). All
line numbers in the old inventory had gone stale after the set_disk /
diagnostics / cluster refactors, so this re-scans and reduces the marker
count from 144 to 99.

STALE removals (comment describes already-implemented behavior, or dead
commented-out blocks) across ecstore (set_disk ops/core, store,
cluster/rpc, bucket/metadata_sys, services), iam, filemeta, s3select and
rustfs auth/object_usecase. No behavior change.

Safe fills, each verified:
- filemeta: replication_info_equals now also compares
  replication_state_internal (function currently has no callers; adds a
  regression test).
- bitrot: drop the confirmed-unused `_want` parameter from bitrot_verify
  and the now-unused `sum` on LocalDisk::bitrot_verify, removing a
  Bytes::copy_from_slice allocation. Streaming verify uses the file's
  embedded per-shard hash, never the passed sum.
- signer: rename v4_ignored_headers -> V4_IGNORED_HEADERS and drop the
  non_upper_case_globals allow.
- admin/heal: test_decode was #[ignore]d and used serde_urlencoded on a
  JSON body (would panic); rewire to serde_json::from_slice to match the
  production decode path, add assertions, un-ignore.

Verified: cargo fmt; cargo check on touched crates; tests pass
(filemeta, signer, bitrot, heal::test_decode); arch guardrail scripts
pass.
This commit is contained in:
Zhengchao An
2026-07-06 22:43:32 +08:00
committed by GitHub
parent 5f1759eb3c
commit 31c6859965
27 changed files with 58 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/metadata_sys.rs
+1 -3
View File
@@ -295,7 +295,6 @@ impl BucketMetadataSys {
let mut futures = Vec::new();
for bucket in buckets.iter() {
// TODO: HealBucket
let api = self.api.clone();
let bucket = bucket.clone();
futures.push(async move {
@@ -319,14 +318,13 @@ impl BucketMetadataSys {
let mut mp = self.metadata_map.write().await;
// TODO:EventNotifier,BucketTargetSys
// TODO: EventNotifier
for res in results {
match res {
Ok(res) => {
if let Some(bucket) = buckets.get(idx) {
let x = Arc::new(res);
mp.insert(bucket.clone(), x.clone());
// TODO:EventNotifier,BucketTargetSys
BucketTargetSys::get().set(bucket, &x).await;
}
}
crates/ecstore/src/cluster/rpc/remote_disk.rs
-51
View File
@@ -918,7 +918,6 @@ fn decode_batch_read_version_response_items(
Ok(batch_read_version_resps)
}
// TODO: all api need to handle errors
#[async_trait::async_trait]
impl DiskAPI for RemoteDisk {
#[tracing::instrument(skip(self))]
@@ -1185,56 +1184,6 @@ impl DiskAPI for RemoteDisk {
.await
}
// // FIXME: TODO: use writer
// #[tracing::instrument(skip(self, wr))]
// async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
// let now = std::time::SystemTime::now();
// info!("walk_dir {}/{}/{:?}", self.endpoint.to_string(), opts.bucket, opts.filter_prefix);
// let mut wr = wr;
// let mut out = MetacacheWriter::new(&mut wr);
// let mut buf = Vec::new();
// opts.serialize(&mut Serializer::new(&mut buf))?;
// let mut client = node_service_time_out_client(&self.addr)
// .await
// .map_err(|err| Error::other(format!("can not get client, err: {}", err)))?;
// let request = Request::new(WalkDirRequest {
// disk: self.endpoint.to_string(),
// walk_dir_options: buf.into(),
// });
// let mut response = client.walk_dir(request).await?.into_inner();
// loop {
// match response.next().await {
// Some(Ok(resp)) => {
// if !resp.success {
// if let Some(err) = resp.error_info {
// if err == "Unexpected EOF" {
// return Err(Error::Io(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, err)));
// } else {
// return Err(Error::other(err));
// }
// }
// return Err(Error::other("unknown error"));
// }
// let entry = serde_json::from_str::<MetaCacheEntry>(&resp.meta_cache_entry)
// .map_err(|_| Error::other(format!("Unexpected response: {:?}", response)))?;
// out.write_obj(&entry).await?;
// }
// None => break,
// _ => return Err(Error::other(format!("Unexpected response: {:?}", response))),
// }
// }
// info!(
// "walk_dir {}/{:?} done {:?}",
// opts.bucket,
// opts.filter_prefix,
// now.elapsed().unwrap_or_default()
// );
// Ok(())
// }
#[tracing::instrument(skip(self))]
async fn delete_version(
&self,
crates/ecstore/src/core/sets.rs
-46
View File
@@ -176,7 +176,6 @@ impl Sets {
let sets = Arc::new(Self {
id: fm.id,
// sets: todo!(),
disk_set,
pool_idx,
endpoints: endpoints.clone(),
@@ -194,10 +193,6 @@ impl Sets {
let rx1 = rx.resubscribe();
tokio::spawn(async move { asets.monitor_and_connect_endpoints(rx1).await });
// let sets2 = sets.clone();
// let rx2 = rx.resubscribe();
// tokio::spawn(async move { sets2.cleanup_deleted_objects_loop(rx2).await });
Ok(sets)
}
@@ -205,36 +200,6 @@ impl Sets {
self.set_drive_count
}
// pub async fn cleanup_deleted_objects_loop(self: Arc<Self>, mut rx: Receiver<()>) {
// tokio::time::sleep(Duration::from_secs(5)).await;
// info!("start cleanup_deleted_objects_loop");
// // TODO: config interval
// let mut interval = tokio::time::interval(Duration::from_secs(15 * 3));
// loop {
// tokio::select! {
// _= interval.tick()=>{
// info!("cleanup_deleted_objects_loop tick");
// for set in self.disk_set.iter() {
// set.clone().cleanup_deleted_objects().await;
// }
// interval.reset();
// },
// _ = rx.recv() => {
// warn!("cleanup_deleted_objects_loop ctx cancelled");
// break;
// }
// }
// }
// warn!("cleanup_deleted_objects_loop exit");
// }
pub async fn monitor_and_connect_endpoints(&self, mut rx: Receiver<()>) {
tokio::time::sleep(Duration::from_secs(5)).await;
@@ -327,17 +292,6 @@ impl Sets {
}
}
// async fn commit_rename_data_dir(
// &self,
// disks: &Vec<Option<DiskStore>>,
// bucket: &str,
// object: &str,
// data_dir: &str,
// // write_quorum: usize,
// ) -> Vec<Option<Error>> {
// unimplemented!()
// }
async fn delete_prefix(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> {
let mut futures = Vec::new();
let mut opt = opts.clone();
crates/ecstore/src/disk/local.rs
+2 -19
View File
@@ -1785,14 +1785,7 @@ impl LocalDisk {
DiskMetrics::default()
}
async fn bitrot_verify(
&self,
part_path: &PathBuf,
part_size: usize,
algo: HashAlgorithm,
sum: &[u8],
shard_size: usize,
) -> Result<()> {
async fn bitrot_verify(&self, part_path: &PathBuf, part_size: usize, algo: HashAlgorithm, shard_size: usize) -> Result<()> {
let retry_count = bitrot_size_mismatch_retry_count();
let retry_delay = bitrot_size_mismatch_retry_delay();
@@ -1801,16 +1794,7 @@ impl LocalDisk {
let meta = file.metadata().await.map_err(to_file_error)?;
let file_size = meta.len() as usize;
match bitrot_verify(
Box::new(file),
file_size,
part_size,
algo.clone(),
Bytes::copy_from_slice(sum),
shard_size,
)
.await
{
match bitrot_verify(Box::new(file), file_size, part_size, algo.clone(), shard_size).await {
Ok(()) => return Ok(()),
Err(err) if attempt < retry_count && is_bitrot_size_mismatch_error(&err) => {
info!(
@@ -2622,7 +2606,6 @@ impl DiskAPI for LocalDisk {
&part_path,
erasure.shard_file_size(part.size as i64) as usize,
checksum_algo,
&checksum_info.hash,
erasure.shard_size(),
)
.await
crates/ecstore/src/erasure/coding/bitrot.rs
-2
View File
@@ -12,7 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use bytes::Bytes;
use pin_project_lite::pin_project;
use rustfs_utils::HashAlgorithm;
use std::io::IoSlice;
@@ -221,7 +220,6 @@ pub async fn bitrot_verify<R: AsyncRead + Unpin + Send>(
want_size: usize,
part_size: usize,
algo: HashAlgorithm,
_want: Bytes, // FIXME: useless parameter?
mut shard_size: usize,
) -> std::io::Result<()> {
let mut hash_buf = vec![0; algo.size()];
crates/ecstore/src/object_api/types.rs
-3
View File
@@ -365,7 +365,6 @@ impl ObjectInfo {
if let Some(size_str) = rustfs_utils::http::get_str(&self.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE)
&& !size_str.is_empty()
{
// Todo: deal with error
let size = size_str.parse::<i64>().map_err(|e| std::io::Error::other(e.to_string()))?;
return Ok(size);
}
@@ -509,8 +508,6 @@ impl ObjectInfo {
})
.collect::<Vec<_>>();
// TODO: part checksums
ObjectInfo {
bucket: bucket.to_string(),
name,
crates/ecstore/src/services/event_notification.rs
-5
View File
@@ -52,11 +52,6 @@ impl EventNotifier {
}
fn init_bucket_targets(&self, _api: ECStore) -> Result<(), std::io::Error> {
/*if err := self.target_list.Add(globalNotifyTargetList.Targets()...); err != nil {
return err
}
self.target_list = self.target_list.Init(runtime.GOMAXPROCS(0)) // TODO: make this configurable (y4m4)
nil*/
warn!("init_bucket_targets called but currently no-op in this build");
Ok(())
}
crates/ecstore/src/services/rebalance/entry.rs
-5
View File
@@ -356,11 +356,6 @@ impl ECStore {
"Rebalance bucket entry scan started"
);
// TODO: other config
// if bucket != RUSTFS_META_BUCKET{
// }
let pool = clone_arc_by_index(self.pools.as_slice(), pool_index, "invalid rebalance pool index")?;
let bucket_configs = Arc::new(load_rebalance_bucket_configs(&bucket).await?);
crates/ecstore/src/set_disk/core/io_primitives.rs
-14
View File
@@ -2476,7 +2476,6 @@ impl SetDisks {
let mut futures = Vec::with_capacity(disks.len());
if let Some(ret_err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
// TODO: add concurrency
for (i, err) in errs.iter().enumerate() {
if err.is_some() {
continue;
@@ -2562,19 +2561,6 @@ impl SetDisks {
vec![None; disks.len()]
};
// // TODO: reduce_common_data_dir
// if let Some(old_dir) = rename_ress
// .iter()
// .filter_map(|v| if v.is_some() { v.as_ref().unwrap().old_data_dir } else { None })
// .map(|v| v.to_string())
// .next()
// {
// let cm_errs = self.commit_rename_data_dir(&shuffle_disks, &bucket, &object, &old_dir).await;
// warn!("put_object commit_rename_data_dir:{:?}", &cm_errs);
// }
// self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await?;
Ok((online_disks, versions, data_dir, cleanup_disks))
}
crates/ecstore/src/set_disk/mod.rs
-8
View File
@@ -1901,14 +1901,6 @@ impl SetDisks {
// Ok(())
// }
// Shuffle the order
// Shuffle the order
// Return shuffled partsMetadata depending on distribution.
// shuffle_disks TODO: use origin value
}
fn is_explicit_null_version(version_id: Option<Uuid>) -> bool {
crates/ecstore/src/set_disk/ops/multipart.rs
-2
View File
@@ -1181,8 +1181,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
checksum_combined.extend_from_slice(cs.raw.as_slice());
}
// TODO: check min part size
object_size += ext_part.size;
object_actual_size += ext_part.actual_size;
crates/ecstore/src/set_disk/ops/object.rs
-7
View File
@@ -783,8 +783,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
.try_get_index()
.map(crate::io_support::rio::compression_index_storage_bytes);
//TODO: userDefined
let mut etag = data.stream.try_resolve_etag().unwrap_or_default();
if let Some(ref tag) = opts.preserve_etag {
etag = tag.clone();
@@ -1726,8 +1724,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
self.invalidate_get_object_metadata_cache(bucket, object).await;
// TODO: nslock
// Guard lock for metadata update
let _lock_guard = if !opts.no_lock {
Some(self.acquire_write_lock_diag("put_object_metadata", bucket, object).await?)
@@ -2179,11 +2175,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned());
// TODO: userdeefined
self.update_object_meta(bucket, object, fi.clone(), disks.as_slice()).await?;
// TODO: versioned
Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
}
crates/ecstore/src/store/list_objects.rs
-14
View File
@@ -1542,12 +1542,6 @@ impl ECStore {
.instrument(tracing::Span::current()),
);
// let merge_res = merge_entry_channels(rx, inputs, sender.clone(), 1).await;
// TODO: cancelList
// let merge_res = merge_entry_channels(rx, inputs, sender.clone(), 1).await;
let results = join_all(futures).await;
let mut all_at_eof = true;
@@ -1572,10 +1566,6 @@ impl ECStore {
return Ok(Vec::new());
}
// merge_res?;
// TODO check cancel
for err in errs.iter() {
if let Some(err) = err {
if err == &Error::Unexpected {
@@ -2010,9 +2000,6 @@ async fn gather_results(
entry.name = entry.name.replace("\\", "/");
}
// TODO: rx.recv()
// TODO: isLatestDeletemarker
if !opts.include_directories
&& (entry.is_dir() || (!opts.versioned && entry.is_object() && entry.is_latest_delete_marker()))
{
@@ -2212,7 +2199,6 @@ async fn merge_entry_channels(
if !dir_matches {
// dir and object has the save name
if other_entry.is_dir() {
// TODO: read next entry to top
if !select_from(&rx, &mut in_channels, other_idx, &mut top, &mut n_done).await? {
return Ok(());
}
crates/ecstore/src/store/mod.rs
+1 -2
View File
@@ -407,7 +407,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
self.handle_verify_object_integrity(bucket, object, opts).await
}
// TODO: review
#[instrument(skip(self))]
async fn copy_object(
&self,
@@ -436,7 +435,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
self.handle_delete_object(bucket, object, opts).await
}
// TODO: review
#[instrument(skip(self))]
async fn delete_objects(
&self,