mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 20:06:37 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8b08c497e9 |
@@ -866,7 +866,7 @@ impl BucketTargetSys {
|
|||||||
return Some(cli);
|
return Some(cli);
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: spawn a task to reload the target
|
// TODO(backlog): spawn an async task to proactively reload the replication target
|
||||||
if self.is_reloading_target(bucket, arn).await {
|
if self.is_reloading_target(bucket, arn).await {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -454,7 +454,7 @@ impl S3PeerSys {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
topology_complete &= bucket_map.values().all(|count| *count >= quorum);
|
topology_complete &= bucket_map.values().all(|count| *count >= quorum);
|
||||||
// TODO: MRF
|
// TODO(backlog): integrate MRF backlog stats into scanner bucket listing
|
||||||
}
|
}
|
||||||
|
|
||||||
let mut buckets: Vec<BucketInfo> = result_map.into_values().collect();
|
let mut buckets: Vec<BucketInfo> = result_map.into_values().collect();
|
||||||
|
|||||||
@@ -2406,7 +2406,7 @@ impl DiskAPI for RemoteDisk {
|
|||||||
return errors;
|
return errors;
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: use Error not string
|
// TODO(backlog): replace string errors with typed `StorageError` variants
|
||||||
|
|
||||||
let result = self
|
let result = self
|
||||||
.execute_with_timeout(
|
.execute_with_timeout(
|
||||||
|
|||||||
@@ -249,7 +249,7 @@ impl Sets {
|
|||||||
|
|
||||||
self.connect_disks().await;
|
self.connect_disks().await;
|
||||||
|
|
||||||
// TODO: config interval
|
// TODO(backlog): make monitor_and_connect interval configurable instead of hardcoded 15s
|
||||||
let mut interval = tokio::time::interval(Duration::from_secs(15));
|
let mut interval = tokio::time::interval(Duration::from_secs(15));
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
|
|||||||
@@ -5215,8 +5215,8 @@ impl LocalDisk {
|
|||||||
|
|
||||||
let cache = Cache::new(update_fn, Duration::from_secs(1), Opts::default());
|
let cache = Cache::new(update_fn, Duration::from_secs(1), Opts::default());
|
||||||
|
|
||||||
// TODO: DIRECT support
|
// TODO(backlog): add O_DIRECT I/O support for performance-critical paths
|
||||||
// TODD: DiskInfo
|
// TODO(backlog): populate DiskInfo in constructor
|
||||||
let mut disk = Self {
|
let mut disk = Self {
|
||||||
root: root.clone(),
|
root: root.clone(),
|
||||||
publication_root,
|
publication_root,
|
||||||
@@ -5751,7 +5751,7 @@ impl LocalDisk {
|
|||||||
|
|
||||||
// return Ok(());
|
// return Ok(());
|
||||||
|
|
||||||
// TODO: async notifications for disk space checks and trash cleanup
|
// TODO(backlog): make disk space checks and trash cleanup event-driven instead of poll-based
|
||||||
|
|
||||||
let trash_path = self.io_get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
|
let trash_path = self.io_get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
|
||||||
// if let Some(parent) = trash_path.parent() {
|
// if let Some(parent) = trash_path.parent() {
|
||||||
@@ -5997,7 +5997,7 @@ impl LocalDisk {
|
|||||||
|
|
||||||
#[hotpath::measure(impl_type = "LocalDisk")]
|
#[hotpath::measure(impl_type = "LocalDisk")]
|
||||||
async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
|
async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
|
||||||
// TODO: timeout support
|
// TODO(backlog): add configurable timeout for read_all_data operations
|
||||||
let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
|
let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
|
||||||
Ok(data)
|
Ok(data)
|
||||||
}
|
}
|
||||||
@@ -6674,7 +6674,7 @@ impl LocalDisk {
|
|||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: add lock
|
// TODO(backlog): add directory listing lock to prevent concurrent enumeration
|
||||||
|
|
||||||
let stall = opts.stall_timeout_duration();
|
let stall = opts.stall_timeout_duration();
|
||||||
|
|
||||||
@@ -8796,7 +8796,7 @@ impl DiskAPI for LocalDisk {
|
|||||||
Ok(entries)
|
Ok(entries)
|
||||||
}
|
}
|
||||||
|
|
||||||
// FIXME: TODO: io.writer TODO cancel
|
// TODO(backlog): support io.writer cancellation and early termination in walk_dir
|
||||||
#[tracing::instrument(level = "trace", skip_all)]
|
#[tracing::instrument(level = "trace", skip_all)]
|
||||||
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
|
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
|
||||||
self.wait_for_startup_cleanup().await;
|
self.wait_for_startup_cleanup().await;
|
||||||
@@ -9880,7 +9880,7 @@ impl DiskAPI for LocalDisk {
|
|||||||
);
|
);
|
||||||
return Err(e);
|
return Err(e);
|
||||||
}
|
}
|
||||||
// TODO: health check
|
// TODO(backlog): add post-setup disk health verification
|
||||||
}
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -249,7 +249,7 @@ impl PoolEndpointList {
|
|||||||
endpoint.set_set_index(0);
|
endpoint.set_set_index(0);
|
||||||
endpoint.set_disk_index(0);
|
endpoint.set_disk_index(0);
|
||||||
|
|
||||||
// TODO Check for cross device mounts if any.
|
// TODO(backlog): check for cross-device mounts in single-drive setup
|
||||||
|
|
||||||
return Ok(Self {
|
return Ok(Self {
|
||||||
inner: vec![Endpoints::from(vec![endpoint])],
|
inner: vec![Endpoints::from(vec![endpoint])],
|
||||||
@@ -264,7 +264,7 @@ impl PoolEndpointList {
|
|||||||
// Convert args to endpoints
|
// Convert args to endpoints
|
||||||
let mut eps = Endpoints::try_from(set_layout.as_slice())?;
|
let mut eps = Endpoints::try_from(set_layout.as_slice())?;
|
||||||
|
|
||||||
// TODO Check for cross device mounts if any.
|
// TODO(backlog): check for cross-device mounts in multi-pool setup
|
||||||
|
|
||||||
for (disk_idx, ep) in eps.as_mut().iter_mut().enumerate() {
|
for (disk_idx, ep) in eps.as_mut().iter_mut().enumerate() {
|
||||||
ep.set_pool_index(pool_idx);
|
ep.set_pool_index(pool_idx);
|
||||||
|
|||||||
@@ -1091,7 +1091,7 @@ impl ObjectInfo {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
// TODO:VersionPurgeStatus
|
// TODO(backlog): handle VersionPurgeStatus in object listing
|
||||||
let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
|
let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
|
||||||
objects.push(ObjectInfo::from_file_info(&fi, bucket, &entry.name, versioned));
|
objects.push(ObjectInfo::from_file_info(&fi, bucket, &entry.name, versioned));
|
||||||
|
|
||||||
|
|||||||
@@ -1575,7 +1575,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
|||||||
let parts_metadata = vec![fi.clone(); disks.len()];
|
let parts_metadata = vec![fi.clone(); disks.len()];
|
||||||
|
|
||||||
if !user_defined.contains_key("content-type") {
|
if !user_defined.contains_key("content-type") {
|
||||||
// TODO: get content-type
|
// TODO(backlog): detect content-type from part data when header is missing
|
||||||
}
|
}
|
||||||
|
|
||||||
if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS)
|
if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS)
|
||||||
@@ -1971,7 +1971,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
|||||||
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
|
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: crypto
|
// TODO(backlog): integrate encryption verification during complete multipart
|
||||||
|
|
||||||
if (i < uploaded_parts.len() - 1)
|
if (i < uploaded_parts.len() - 1)
|
||||||
&& !(opts.data_movement && ext_part.actual_size < 0)
|
&& !(opts.data_movement && ext_part.actual_size < 0)
|
||||||
|
|||||||
@@ -6161,7 +6161,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
|||||||
|
|
||||||
join_all(rollback_futures).await;
|
join_all(rollback_futures).await;
|
||||||
|
|
||||||
// TODO: add_partial
|
// TODO(backlog): support partial object deletion for multi-part objects
|
||||||
|
|
||||||
if let Some(api) = opts.tier_delete_journal_api.as_ref() {
|
if let Some(api) = opts.tier_delete_journal_api.as_ref() {
|
||||||
for (idx, je) in persisted_journal_entries {
|
for (idx, je) in persisted_journal_entries {
|
||||||
@@ -6371,7 +6371,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: Lifecycle
|
// TODO(backlog): integrate lifecycle evaluation before object deletion
|
||||||
|
|
||||||
let mut version_found = true;
|
let mut version_found = true;
|
||||||
// delete_object_version below derives its own majority quorum from the
|
// delete_object_version below derives its own majority quorum from the
|
||||||
@@ -6465,7 +6465,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
|||||||
mark_deleted: mark_delete,
|
mark_deleted: mark_delete,
|
||||||
mod_time: Some(mod_time),
|
mod_time: Some(mod_time),
|
||||||
replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta),
|
replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta),
|
||||||
..Default::default() // TODO: Transition
|
..Default::default() // TODO(backlog): populate transition state on delete markers
|
||||||
};
|
};
|
||||||
|
|
||||||
fi.set_tier_free_version_id(&find_vid.to_string());
|
fi.set_tier_free_version_id(&find_vid.to_string());
|
||||||
|
|||||||
@@ -601,7 +601,7 @@ impl ECStore {
|
|||||||
|
|
||||||
#[instrument(skip(self))]
|
#[instrument(skip(self))]
|
||||||
pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
|
pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
|
||||||
// TODO: opts.cached
|
// TODO(backlog): support cached bucket listing via opts.cached
|
||||||
|
|
||||||
let mut buckets = self.peer_sys.list_bucket(opts).await?;
|
let mut buckets = self.peer_sys.list_bucket(opts).await?;
|
||||||
|
|
||||||
|
|||||||
@@ -4673,7 +4673,7 @@ async fn gather_results(
|
|||||||
entry.name = entry.name.replace("\\", "/");
|
entry.name = entry.name.replace("\\", "/");
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: rx.recv()
|
// TODO(backlog): integrate rx.recv() for incremental listing results
|
||||||
|
|
||||||
if let Some(marker) = &opts.marker
|
if let Some(marker) = &opts.marker
|
||||||
&& ((!opts.include_marker && &entry.name <= marker) || (opts.include_marker && &entry.name < marker))
|
&& ((!opts.include_marker && &entry.name <= marker) || (opts.include_marker && &entry.name < marker))
|
||||||
@@ -4703,7 +4703,7 @@ async fn gather_results(
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: Lifecycle
|
// TODO(backlog): integrate lifecycle evaluation during object listing
|
||||||
|
|
||||||
entries.push(Some(entry));
|
entries.push(Some(entry));
|
||||||
candidate_entries += 1;
|
candidate_entries += 1;
|
||||||
|
|||||||
@@ -332,7 +332,7 @@ impl ECStore {
|
|||||||
let expected_incarnation_id = opts.expected_bucket_incarnation_id;
|
let expected_incarnation_id = opts.expected_bucket_incarnation_id;
|
||||||
|
|
||||||
if request.prefix.is_empty() {
|
if request.prefix.is_empty() {
|
||||||
// TODO: return from cache
|
// TODO(backlog): return cached multipart listing when prefix is empty
|
||||||
}
|
}
|
||||||
|
|
||||||
if self.single_pool() {
|
if self.single_pool() {
|
||||||
@@ -610,7 +610,7 @@ impl ECStore {
|
|||||||
let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
|
let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
|
||||||
let opts = &opts;
|
let opts = &opts;
|
||||||
|
|
||||||
// TODO: defer DeleteUploadID
|
// TODO(backlog): defer DeleteUploadID to background for faster abort response
|
||||||
|
|
||||||
if self.single_pool() {
|
if self.single_pool() {
|
||||||
return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
|
return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
|
||||||
|
|||||||
@@ -385,7 +385,7 @@ impl ECStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub(super) async fn is_suspended(&self, idx: usize) -> bool {
|
pub(super) async fn is_suspended(&self, idx: usize) -> bool {
|
||||||
// TODO: LOCK
|
// TODO(backlog): acquire pool metadata lock for consistent suspension check
|
||||||
|
|
||||||
let pool_meta = self.pool_meta.read().await;
|
let pool_meta = self.pool_meta.read().await;
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user