diff --git a/crates/ecstore/benches/single_block_non_inline_benchmark.rs b/crates/ecstore/benches/single_block_non_inline_benchmark.rs index 711bbef27..7f784ee94 100644 --- a/crates/ecstore/benches/single_block_non_inline_benchmark.rs +++ b/crates/ecstore/benches/single_block_non_inline_benchmark.rs @@ -69,6 +69,7 @@ fn build_non_inline_writers(config: &BenchConfig) -> Vec bool { }) } +fn small_ingest_capacity(erasure: &Erasure, size_hint: usize) -> usize { + let data_len = size_hint.min(erasure.block_size); + erasure.encoded_capacity_for_data_len(data_len).min(erasure.block_size) +} + /// Keeps the encoder producer scoped to its parent future. Tokio detaches a /// task when its `JoinHandle` is dropped, so the producer must be aborted when /// an upload is cancelled before the encode pipeline finishes. @@ -540,13 +545,14 @@ impl Erasure { writers: &mut [Option], quorum: usize, require_single_block: bool, + size_hint: usize, ) -> std::io::Result<(R, usize)> where R: AsyncRead + Send + Sync + Unpin, { use tokio::io::AsyncReadExt; - let mut buf = Vec::with_capacity(self.block_size); + let mut buf = Vec::with_capacity(small_ingest_capacity(&self, size_hint)); let total = if require_single_block { let read_limit = self .block_size @@ -880,7 +886,24 @@ impl Erasure { where R: AsyncRead + Send + Sync + Unpin, { - self.encode_small_direct(reader, writers, quorum, false).await + let size_hint = self.block_size; + self.encode_small_direct(reader, writers, quorum, false, size_hint).await + } + + /// Size-aware inline fast path. `size_hint` only controls the bounded initial + /// allocation; reads remain authoritative. + #[hotpath::measure(impl_type = "Erasure")] + pub async fn encode_inline_small_with_size_hint( + self: Arc, + reader: R, + writers: &mut [Option], + quorum: usize, + size_hint: usize, + ) -> std::io::Result<(R, usize)> + where + R: AsyncRead + Send + Sync + Unpin, + { + self.encode_small_direct(reader, writers, quorum, false, size_hint).await } /// Fast path for single-block non-inline objects: avoids the producer/consumer @@ -895,7 +918,24 @@ impl Erasure { where R: AsyncRead + Send + Sync + Unpin, { - self.encode_small_direct(reader, writers, quorum, true).await + let size_hint = self.block_size; + self.encode_small_direct(reader, writers, quorum, true, size_hint).await + } + + /// Size-aware single-block fast path. `size_hint` only controls the bounded + /// initial allocation; reads remain authoritative. + #[hotpath::measure(impl_type = "Erasure")] + pub async fn encode_single_block_non_inline_with_size_hint( + self: Arc, + reader: R, + writers: &mut [Option], + quorum: usize, + size_hint: usize, + ) -> std::io::Result<(R, usize)> + where + R: AsyncRead + Send + Sync + Unpin, + { + self.encode_small_direct(reader, writers, quorum, true, size_hint).await } } @@ -2293,7 +2333,10 @@ mod tests { let erasure = Arc::new(Erasure::new(1, 0, 16)); let reader = tokio::io::BufReader::new(Cursor::new(Vec::::new())); - let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, 1).await.unwrap(); + let (_reader, total) = erasure + .encode_inline_small_with_size_hint(reader, &mut writers, 1, 0) + .await + .unwrap(); assert_eq!(total, 0); // No shutdown was called, so nothing should be committed @@ -2325,7 +2368,10 @@ mod tests { let payload = b"hello inline small"; let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec())); - let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, DATA_SHARDS).await.unwrap(); + let (_reader, total) = erasure + .encode_inline_small_with_size_hint(reader, &mut writers, DATA_SHARDS, 1) + .await + .unwrap(); assert_eq!(total, payload.len()); // All shards must have received data (shutdown flushed the bitrot header + shard bytes) @@ -2392,7 +2438,7 @@ mod tests { let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); let reader = tokio::io::BufReader::new(Cursor::new(payload)); let err = erasure - .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS) + .encode_single_block_non_inline_with_size_hint(reader, &mut writers, DATA_SHARDS, BLOCK_SIZE) .await .expect_err("single-block fast path must reject oversized readers"); @@ -2403,6 +2449,21 @@ mod tests { } } + #[test] + fn small_ingest_capacity_uses_bounded_size_hint() { + let erasure = Erasure::new(4, 2, 1024 * 1024); + assert_eq!(small_ingest_capacity(&erasure, 0), 0); + assert_eq!(small_ingest_capacity(&erasure, 4 * 1024), 6 * 1024); + assert_eq!(small_ingest_capacity(&erasure, 16 * 1024), 24 * 1024); + assert_eq!(small_ingest_capacity(&erasure, usize::MAX), 1024 * 1024); + + let legacy = Erasure::new_with_options(4, 2, 1024 * 1024, true); + assert_eq!(small_ingest_capacity(&legacy, 4 * 1024), 6 * 1024); + + let high_parity = Erasure::new(4, 12, 1024 * 1024); + assert_eq!(small_ingest_capacity(&high_parity, usize::MAX), 1024 * 1024); + } + #[tokio::test] async fn read_full_buf_or_eof_returns_none_on_empty_reader() { let mut reader = Cursor::new(Vec::::new()); diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index f80bd0570..8b4e213b2 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -968,6 +968,15 @@ impl Erasure { self.data_shards + self.parity_shards } + pub(crate) fn encoded_capacity_for_data_len(&self, data_len: usize) -> usize { + let shard_size_fn = if self.uses_legacy { + calc_shard_size_legacy + } else { + calc_shard_size + }; + shard_size_fn(data_len, self.data_shards).saturating_mul(self.total_shard_count()) + } + /// Whether the erasure dimensions are safe for the shard/offset arithmetic. /// /// `block_size` and `data_shards` come straight from on-disk metadata; a diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 4f1e4c2f3..a6c3095e0 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -58,6 +58,7 @@ use crate::io_support::bitrot::{ create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, object_mmap_read_max_length, }; use crate::set_disk::shard_source::ShardReadCost; +use futures::FutureExt as _; use futures::stream::{FuturesUnordered, StreamExt}; use metrics::counter; use std::{ @@ -2856,8 +2857,6 @@ impl SetDisks { file_info.validate_for_erasure_write()?; } } - let mut futures = Vec::with_capacity(disks.len()); - let mut errs = Vec::with_capacity(disks.len()); let src_bucket = Arc::new(src_bucket.to_string()); @@ -2865,48 +2864,65 @@ impl SetDisks { let dst_bucket = Arc::new(dst_bucket.to_string()); let dst_object = Arc::new(dst_object.to_string()); - for (i, (disk, file_info)) in disks.iter().zip(file_infos.iter()).enumerate() { - let mut file_info = file_info.clone(); - let disk = disk.clone(); - let src_bucket = src_bucket.clone(); - let src_object = src_object.clone(); - let dst_object = dst_object.clone(); - let dst_bucket = dst_bucket.clone(); + let disk_count = disks.len(); + let fanout_disks = disks.to_vec(); + let fanout_file_infos = file_infos.to_vec(); + let fanout_src_bucket = src_bucket.clone(); + let fanout_src_object = src_object.clone(); + let fanout_dst_bucket = dst_bucket.clone(); + let fanout_dst_object = dst_object.clone(); + // Keep one coordinator task so a cancelled caller cannot drop partially + // completed disk mutations. Per-disk futures stay ordered in `join_all`, + // preserving slot-indexed quorum and convergence accounting without a + // scheduler task for every disk. + let fanout = tokio::spawn(async move { + let futures = fanout_disks + .into_iter() + .zip(fanout_file_infos) + .enumerate() + .map(|(i, (disk, mut file_info))| { + let src_bucket = fanout_src_bucket.clone(); + let src_object = fanout_src_object.clone(); + let dst_object = fanout_dst_object.clone(); + let dst_bucket = fanout_dst_bucket.clone(); - futures.push(tokio::spawn(async move { - // Test-only introspection guard: counts this task as in-flight for - // the whole body. Compiles to `()` in production (no behavior). - #[allow(clippy::let_unit_value)] - let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object); + std::panic::AssertUnwindSafe(async move { + // Test-only introspection guard: counts this operation as + // in-flight for the whole body. Compiles to `()` in production. + #[allow(clippy::let_unit_value)] + let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object); - let Some(disk) = disk else { - return Err(DiskError::DiskNotFound); - }; + let Some(disk) = disk else { + return Err(DiskError::DiskNotFound); + }; - let is_delete_marker = file_info.is_canonical_delete_marker(); - if file_info.erasure.index == 0 { - file_info.erasure.index = i + 1; - } + let is_delete_marker = file_info.is_canonical_delete_marker(); + if file_info.erasure.index == 0 { + file_info.erasure.index = i + 1; + } - if !is_delete_marker && !file_info.has_valid_erasure_geometry() { - return Err(DiskError::FileCorrupt); - } + if !is_delete_marker && !file_info.has_valid_erasure_geometry() { + return Err(DiskError::FileCorrupt); + } - // Test-only awaitable pause point right before the disk rename. - // A no-op immediately-ready future in production. - Self::rename_fanout_barrier(&dst_object, i, rename_fanout_barrier_phase::RENAME).await; + // Test-only awaitable pause point right before the disk rename. + // A no-op immediately-ready future in production. + Self::rename_fanout_barrier(&dst_object, i, rename_fanout_barrier_phase::RENAME).await; - disk.rename_data(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object) - .await - })); - } + disk.rename_data(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object) + .await + }) + .catch_unwind() + }); + join_all(futures).await + }); - let mut disk_versions = vec![None; disks.len()]; - let mut data_dirs = vec![None; disks.len()]; - let mut cleanup_data_dirs = vec![None; disks.len()]; - let mut old_current_sizes = vec![None; disks.len()]; + let mut disk_versions = vec![None; disk_count]; + let mut data_dirs = vec![None; disk_count]; + let mut cleanup_data_dirs = vec![None; disk_count]; + let mut old_current_sizes = vec![None; disk_count]; - let results = join_all(futures).await; + let results = fanout.await.map_err(|_| DiskError::Unexpected)?; for (idx, result) in results.iter().enumerate() { match result.as_ref().map_err(|_| DiskError::Unexpected)? { @@ -5867,6 +5883,51 @@ mod tests { drop(dirs); } + #[tokio::test] + async fn rename_fanout_drains_after_caller_cancellation() { + const DISKS: usize = 4; + let bucket = "rename-cancel-bucket"; + let object = "rename-cancel-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + let marker = metadata_test_delete_marker(object, Uuid::new_v4(), OffsetDateTime::now_utc()); + let file_infos = vec![marker; DISKS]; + let tracker = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + + let rename = + tokio::spawn( + async move { SetDisks::rename_data(&disks, bucket, object, &file_infos, bucket, object, DISKS - 1).await }, + ); + tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused()) + .await + .expect("rename fan-out must reach the armed barrier"); + rename.abort(); + assert!( + rename + .await + .expect_err("aborted caller should report cancellation") + .is_cancelled(), + "caller task should be cancelled, not panic" + ); + assert!(tracker.running() >= 1, "the coordinator must retain in-flight disk mutations"); + + barrier.release(); + tokio::time::timeout(BARRIER_PAUSE_GUARD, async { + while tracker.running() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("cancelled caller's disk mutations must drain"); + + for (idx, dir) in dirs.iter().enumerate() { + assert!( + dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE).exists(), + "disk {idx} must finish the rename after caller cancellation" + ); + } + } + /// Demo / regression guard for the barrier on the commit (old-data-dir) /// cleanup fan-out. Serves the same #1312/#1319 "no background disk write /// after release" shape, on the reclamation path that runs *after* a write is diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 0505526ec..7b0746905 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -954,12 +954,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let write_path = classify_multipart_part_write_path(multipart_part_size, fi.erasure.block_size); rustfs_io_metrics::record_put_object_path(write_path.multipart_metric_label()); + let small_size_hint = if matches!(write_path, SmallWritePath::SingleBlockNonInline) { + usize::try_from(multipart_part_size).map_err(Error::other)? + } else { + 0 + }; let encode_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); let (reader, w_size) = match write_path { SmallWritePath::SingleBlockNonInline => { Arc::clone(&erasure) - .encode_single_block_non_inline(stream, &mut writers, write_quorum) + .encode_single_block_non_inline_with_size_hint(stream, &mut writers, write_quorum, small_size_hint) .await? } SmallWritePath::PipelineBatchedLarge => { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 8ae8a7ef3..cfcfe0d7c 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -56,6 +56,22 @@ use http::HeaderValue; use rustfs_utils::path::decode_dir_object; use std::future::Future; +#[inline] +fn duration_millis_f64(duration: std::time::Duration) -> f64 { + duration.as_secs_f64() * 1000.0 +} + +#[cfg(test)] +mod duration_metrics_tests { + use super::duration_millis_f64; + use std::time::Duration; + + #[test] + fn duration_millis_preserves_sub_millisecond_precision() { + assert_eq!(duration_millis_f64(Duration::from_micros(125)), 0.125); + } +} + fn is_restore_control_metadata(key: &str) -> bool { key.eq_ignore_ascii_case(X_AMZ_RESTORE.as_str()) || key.eq_ignore_ascii_case(rustfs_utils::http::headers::AMZ_RESTORE_EXPIRY_DAYS) @@ -1107,8 +1123,12 @@ impl SetDisks { writers.push(w); errors.push(e); } - let writer_setup_ms = writer_setup_stage_start.elapsed().as_millis() as u64; - rustfs_io_metrics::record_put_object_stage_duration("set_disk_writer_setup", writer_setup_ms as f64); + let writer_setup_elapsed = writer_setup_stage_start.elapsed(); + let writer_setup_ms = writer_setup_elapsed.as_millis() as u64; + rustfs_io_metrics::record_put_object_stage_duration( + "set_disk_writer_setup", + duration_millis_f64(writer_setup_elapsed), + ); let nil_count = errors.iter().filter(|&e| e.is_none()).count(); if nil_count < write_quorum { @@ -1138,11 +1158,16 @@ impl SetDisks { let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size); rustfs_io_metrics::record_put_object_path(write_path.metric_label()); + let small_size_hint = if matches!(write_path, SmallWritePath::Inline | SmallWritePath::SingleBlockNonInline) { + usize::try_from(put_object_size).map_err(Error::other)? + } else { + 0 + }; let encode_stage_start = Instant::now(); let (reader, w_size) = match write_path { SmallWritePath::Inline => match Arc::clone(&erasure) - .encode_inline_small(stream, &mut writers, write_quorum) + .encode_inline_small_with_size_hint(stream, &mut writers, write_quorum, small_size_hint) .await { Ok((r, w)) => (r, w), @@ -1152,7 +1177,7 @@ impl SetDisks { } }, SmallWritePath::SingleBlockNonInline => match Arc::clone(&erasure) - .encode_single_block_non_inline(stream, &mut writers, write_quorum) + .encode_single_block_non_inline_with_size_hint(stream, &mut writers, write_quorum, small_size_hint) .await { Ok((r, w)) => (r, w), @@ -1178,8 +1203,9 @@ impl SetDisks { } }, }; - let encode_ms = encode_stage_start.elapsed().as_millis() as u64; - rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", encode_ms as f64); + let encode_elapsed = encode_stage_start.elapsed(); + let encode_ms = encode_elapsed.as_millis() as u64; + rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", duration_millis_f64(encode_elapsed)); let _ = mem::replace(&mut data.stream, reader); // if let Err(err) = close_bitrot_writers(&mut writers).await { @@ -1497,8 +1523,9 @@ impl SetDisks { let _ = rustfs_common::heal_channel::send_heal_request(request).await; }); } - let rename_stage_ms = rename_stage_start.elapsed().as_millis() as u64; - rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", rename_stage_ms as f64); + let rename_stage_elapsed = rename_stage_start.elapsed(); + let rename_stage_ms = rename_stage_elapsed.as_millis() as u64; + rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed)); if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { warn!( event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, @@ -1527,9 +1554,13 @@ impl SetDisks { let cleanup = self .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum) .await; - let cleanup_ms = cleanup_stage_start.elapsed().as_millis() as u64; + let cleanup_elapsed = cleanup_stage_start.elapsed(); + let cleanup_ms = cleanup_elapsed.as_millis() as u64; cleanup_stage_ms = Some(cleanup_ms); - rustfs_io_metrics::record_put_object_stage_duration("set_disk_old_data_cleanup", cleanup_ms as f64); + rustfs_io_metrics::record_put_object_stage_duration( + "set_disk_old_data_cleanup", + duration_millis_f64(cleanup_elapsed), + ); self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup) .await; if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index d32d4789d..a826d4a31 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -2513,7 +2513,7 @@ mod tests { json_field: "rename_data_resp", bin_field: "rename_data_resp_bin", }, - json_encoder: "let rename_data_resp_json = compat_response_json(&rename_data_resp, false);", + json_encoder: "let rename_data_resp_json = compat_response_json(rename_data_resp, request_decoded_from_msgpack)", }, ]; diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 37bb0ef14..0e0f2defe 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -18,7 +18,7 @@ use crate::storage::storage_api::rpc_consumer::node_service::{ ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count, }; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; -use crate::storage::storage_api::{PartTransactionAction, SnapshotLeaseToken, verify_tonic_mutation_body_digest}; +use crate::storage::storage_api::{PartTransactionAction, RenameDataResp, SnapshotLeaseToken, verify_tonic_mutation_body_digest}; use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ @@ -214,6 +214,23 @@ fn encode_batch_read_version_response_payloads( Ok((batch_read_version_resps_json, batch_read_version_resps_bin)) } +fn decode_rename_data_request_file_info( + binary: &[u8], + json: &str, +) -> std::result::Result, DiskError> { + decode_msgpack_or_json_with_source(binary, json, "FileInfo") +} + +fn encode_rename_data_response_payloads( + rename_data_resp: &RenameDataResp, + request_decoded_from_msgpack: bool, +) -> std::result::Result<(String, Vec), DiskError> { + let rename_data_resp_json = compat_response_json(rename_data_resp, request_decoded_from_msgpack) + .map_err(|err| DiskError::other(format!("encode RenameDataResp json failed: {err}")))?; + let rename_data_resp_bin = encode_msgpack_named(rename_data_resp, "RenameDataResp")?; + Ok((rename_data_resp_json, rename_data_resp_bin)) +} + impl NodeService { pub(super) async fn handle_acquire_snapshot_lease( &self, @@ -992,7 +1009,7 @@ impl NodeService { )?; let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { + let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) { Ok(file_info) => file_info, Err(err) => { return Ok(Response::new(RenameDataResponse { @@ -1003,31 +1020,30 @@ impl NodeService { })); } }; + let request_decoded_from_msgpack = decoded_file_info.from_msgpack; match disk - .rename_data(&request.src_volume, &request.src_path, file_info, &request.dst_volume, &request.dst_path) + .rename_data( + &request.src_volume, + &request.src_path, + decoded_file_info.value, + &request.dst_volume, + &request.dst_path, + ) .await { Ok(rename_data_resp) => { - let rename_data_resp_json = compat_response_json(&rename_data_resp, false); - let rename_data_resp_bin = encode_msgpack_named(&rename_data_resp, "RenameDataResp"); - match (rename_data_resp_json, rename_data_resp_bin) { - (Ok(rename_data_resp), Ok(rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse { + match encode_rename_data_response_payloads(&rename_data_resp, request_decoded_from_msgpack) { + Ok((rename_data_resp, rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse { success: true, rename_data_resp, rename_data_resp_bin: rename_data_resp_bin.into(), error: None, })), - (Err(err), _) => Ok(Response::new(RenameDataResponse { + Err(err) => Ok(Response::new(RenameDataResponse { success: false, rename_data_resp: String::new(), rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), - })), - (_, Err(err)) => Ok(Response::new(RenameDataResponse { - success: false, - rename_data_resp: String::new(), - rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + error: Some(err.into()), })), } } @@ -1476,12 +1492,14 @@ impl NodeService { #[cfg(test)] mod tests { use super::{ - compat_response_json, decode_msgpack_or_json, encode_batch_read_version_response_payloads, encode_msgpack, - encode_msgpack_named, encode_read_multiple_response_payloads, snapshot_lease_disabled_response, + compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info, + encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named, + encode_read_multiple_response_payloads, encode_rename_data_response_payloads, snapshot_lease_disabled_response, }; - use crate::storage::storage_api::DiskError; use crate::storage::storage_api::ReadMultipleResp; use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp; + use crate::storage::storage_api::{DiskError, RenameDataResp}; + use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::global_internode_metrics; use serde::{Deserialize, Serialize}; @@ -1660,6 +1678,48 @@ mod tests { ); } + #[test] + fn rename_data_response_payloads_follow_successful_request_codec() { + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, None::<&str>), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>), + ], + || { + let response = RenameDataResp::default(); + let file_info = FileInfo::default(); + let legacy_file_info_json = serde_json::to_string(&file_info).expect("FileInfo JSON should encode"); + let legacy_request = decode_rename_data_request_file_info(&[], &legacy_file_info_json) + .expect("legacy FileInfo JSON should decode"); + + let (legacy_json, legacy_bin) = encode_rename_data_response_payloads(&response, legacy_request.from_msgpack) + .expect("legacy response payloads should encode"); + assert!(!legacy_json.is_empty(), "JSON-only requests must retain response JSON"); + assert!(!legacy_bin.is_empty(), "all callers must receive response msgpack"); + + let file_info_bin = encode_file_info_msgpack(&file_info).expect("FileInfo msgpack should encode"); + let msgpack_request = decode_rename_data_request_file_info(&file_info_bin, &legacy_file_info_json) + .expect("FileInfo msgpack should decode"); + let (msgpack_json, msgpack_bin) = encode_rename_data_response_payloads(&response, msgpack_request.from_msgpack) + .expect("msgpack response payloads should encode"); + assert!(msgpack_json.is_empty(), "successfully decoded msgpack requests may omit response JSON"); + let decoded: RenameDataResp = + rmp_serde::from_slice(&msgpack_bin).expect("response msgpack should remain decodable"); + assert_eq!(decoded.old_data_dir, response.old_data_dir); + assert_eq!(decoded.rollback_data_dir, response.rollback_data_dir); + assert_eq!(decoded.cleanup_data_dir, response.cleanup_data_dir); + assert_eq!(decoded.sign, response.sign); + assert_eq!(decoded.old_current_size, response.old_current_size); + + let error = match decode_rename_data_request_file_info(b"not-msgpack", &legacy_file_info_json) { + Ok(_) => panic!("malformed request msgpack must fail closed before response negotiation"), + Err(error) => error, + }; + assert!(error.to_string().contains("decode FileInfo msgpack failed"), "unexpected error: {error}"); + }, + ); + } + #[test] fn decode_msgpack_or_json_fails_closed_on_corrupt_non_empty_msgpack() { let before = global_internode_metrics().msgpack_json_decode_error_total_for_test();