diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 9c10ef078..d1736249c 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -101,7 +101,7 @@ static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new(); async fn finish_resync_workers( worker_txs: Vec>, - results_tx: tokio::sync::broadcast::Sender, + results_tx: tokio::sync::mpsc::Sender, futures: Vec>, abort: bool, ) -> bool { @@ -617,8 +617,12 @@ impl ReplicationResyncer { .unwrap_or_default() }; - let mut last_checkpoint = if status.resync_status == ResyncStatusType::ResyncStarted - || status.resync_status == ResyncStatusType::ResyncFailed + // An empty checkpoint means no per-object progress was persisted before the + // interruption: resume from the beginning, otherwise `object.name != checkpoint` + // below would skip every object and mark the resync completed without work. + let mut last_checkpoint = if (status.resync_status == ResyncStatusType::ResyncStarted + || status.resync_status == ResyncStatusType::ResyncFailed) + && !status.object.is_empty() { Some(status.object) } else { @@ -626,7 +630,9 @@ impl ReplicationResyncer { }; let mut worker_txs = Vec::new(); - let (results_tx, mut results_rx) = tokio::sync::broadcast::channel::(1); + // mpsc, not broadcast: a lagging broadcast receiver returns Err(Lagged) which + // would end the collector and silently drop every subsequent worker result. + let (results_tx, mut results_rx) = tokio::sync::mpsc::channel::(RESYNC_WORKER_COUNT * 4); let opts_clone = opts.clone(); let self_clone = self.clone(); @@ -634,7 +640,7 @@ impl ReplicationResyncer { let mut futures = Vec::new(); let results_fut = tokio::spawn(async move { - while let Ok(st) = results_rx.recv().await { + while let Some(st) = results_rx.recv().await { self_clone.inc_stats(&st, opts_clone.clone()).await; } }); @@ -746,7 +752,7 @@ impl ReplicationResyncer { return; } - if let Err(err) = results_tx.send(st) { + if let Err(err) = results_tx.send(st).await { error!( event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED, component = LOG_COMPONENT_ECSTORE, @@ -2128,7 +2134,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { { Ok(gr) => gr, Err(e) => { - if !is_err_object_not_found(&e) || is_err_version_not_found(&e) { + if !(is_err_object_not_found(&e) || is_err_version_not_found(&e)) { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2414,7 +2420,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { { Ok(gr) => gr, Err(e) => { - if !is_err_object_not_found(&e) || is_err_version_not_found(&e) { + if !(is_err_object_not_found(&e) || is_err_version_not_found(&e)) { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -3376,9 +3382,9 @@ mod tests { #[tokio::test] async fn test_finish_resync_workers_closes_result_collector() { let (worker_tx, mut worker_rx) = tokio::sync::mpsc::channel::(1); - let (results_tx, mut results_rx) = tokio::sync::broadcast::channel::(1); + let (results_tx, mut results_rx) = tokio::sync::mpsc::channel::(1); let worker = tokio::spawn(async move { while worker_rx.recv().await.is_some() {} }); - let collector = tokio::spawn(async move { while results_rx.recv().await.is_ok() {} }); + let collector = tokio::spawn(async move { while results_rx.recv().await.is_some() {} }); let failed = tokio::time::timeout( TokioDuration::from_secs(1), @@ -3392,7 +3398,7 @@ mod tests { #[tokio::test] async fn test_finish_resync_workers_reports_join_failure() { - let (results_tx, _results_rx) = tokio::sync::broadcast::channel::(1); + let (results_tx, _results_rx) = tokio::sync::mpsc::channel::(1); let failed_worker = tokio::spawn(async { panic!("intentional resync worker failure"); }); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index fff424ef8..1150f989f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1623,12 +1623,11 @@ impl LocalDisk { return Ok(()); } - // Update xl.meta + // Update xl.meta atomically: a concurrent reader or crash mid-write must + // never observe a truncated xl.meta for versions that were not deleted. let buf = fm.marshal_msg()?; - let volume_dir = self.get_bucket_path(volume)?; - - self.write_all_private(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), buf.into(), true, &volume_dir) + self.write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &buf, true) .await?; Ok(()) @@ -2873,7 +2872,9 @@ impl DiskAPI for LocalDisk { if let Some(parent) = file_path.parent() { os::make_dir_all(parent, &volume_dir).await?; } - let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY) + // O_TRUNC: if a file already exists at this path, stale trailing bytes past + // the new content would otherwise survive and mismatch the metadata size. + let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY | O_TRUNC) .await .map_err(to_file_error)?; let reclaim_on_shutdown = should_reclaim_file_cache_after_write(_file_size); @@ -3927,7 +3928,9 @@ impl DiskAPI for LocalDisk { let fm_data = meta.marshal_msg()?; - self.write_all(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), fm_data.into()) + // Atomic temp+rename: this path also rewrites live xl.meta (delete markers, + // decommission), where an in-place truncate would expose torn metadata. + self.write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &fm_data, true) .await?; Ok(()) diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 4fffcffa1..793295bfd 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -117,6 +117,10 @@ pub async fn read_dir(path: impl AsRef, count: i32) -> std::io::Result 0 { - self.data_shards + 1 - } else { - self.data_shards - }; - if available_shards < required_shards { - return Err(Error::other(format!( - "can not reconstruct data: not enough verified heal source shards (need {}, have {}) {errs:?}", - required_shards, available_shards - ))); - } - let source_parity = shards .iter() .enumerate() diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index f242463e5..cbbdf5094 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -441,10 +441,11 @@ impl SetDisks { } } - let is_inline_buffer = runtime_sources::storage_class_should_inline( - erasure.shard_file_size(latest_meta.size), - false, - ); + // Preserve the committed layout: recomputing inline-ness here + // (with a hardcoded unversioned threshold) makes healed replicas + // diverge from healthy ones in quorum identity, so heal would + // flag them forever. + let is_inline_buffer = latest_meta.inline_data(); // create writers for all disk positions, but only for outdated disks for (index, disk_op) in out_dated_disks.iter().enumerate() { if let Some(outdated_disk) = disk_op { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 77f205d34..65656867d 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2527,7 +2527,6 @@ impl SetDisks { return Ok(None); } - let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); let checksum_info = fi.erasure.get_checksum_info(part.number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { HashAlgorithm::HighwayHash256SLegacy @@ -2537,7 +2536,11 @@ impl SetDisks { let read_length = erasure.shard_file_offset(0, object_size, object_size); if fi.data.is_some() { - let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(&files, fi, erasure.data_shards, |index| { + // Collect from the canonical disk-ordered inputs: the helper indexes + // fi.erasure.distribution by disk position, so passing shard-ordered + // (shuffled) arrays would apply the permutation twice and concatenate + // the wrong shards into the response body. + let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(files, fi, erasure.data_shards, |index| { disks.get(index).is_some_and(Option::is_some) }) else { return Ok(None); @@ -2587,6 +2590,7 @@ impl SetDisks { return Ok(body); } + let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); let use_mmap_read = object_mmap_read_enabled(); let reader_setup_stage_start = Instant::now(); @@ -2746,6 +2750,15 @@ impl SetDisks { fi.uses_legacy_checksum, ); + // Erasure params come from on-disk metadata; zero values must fail the read + // instead of panicking on the block/shard divisions below. + if erasure.block_size == 0 || erasure.data_shards == 0 { + return Err(Error::other(format!( + "invalid erasure metadata for {bucket}/{object}: block_size={}, data_blocks={}", + erasure.block_size, erasure.data_shards + ))); + } + let part_indices: Vec = (part_index..=last_part_index).collect(); debug!(bucket, object, ?part_indices, "Multipart part indices to stream"); diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index e486be27f..79a83b015 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -665,14 +665,17 @@ impl FileMeta { err = self.add_version_filemata(ventry).err(); } - if self.shared_data_dir_count(obj_version_id, obj_data_dir) > 0 { - return Ok(None); - } - + // A failed delete-marker insertion must surface even when the data dir is + // shared: reporting success here silently turns the delete into a permanent + // delete that replication never propagates. if let Some(e) = err { return Err(e); } + if self.shared_data_dir_count(obj_version_id, obj_data_dir) > 0 { + return Ok(None); + } + Ok(obj_data_dir) } diff --git a/crates/filemeta/src/filemeta/msgp_decode.rs b/crates/filemeta/src/filemeta/msgp_decode.rs index 68e44ca02..b7f9606b6 100644 --- a/crates/filemeta/src/filemeta/msgp_decode.rs +++ b/crates/filemeta/src/filemeta/msgp_decode.rs @@ -179,28 +179,30 @@ pub(crate) fn skip_msgp_value(rd: &mut R) -> Result<()> { } return Ok(()); } - Marker::FixExt1 => 1, - Marker::FixExt2 => 2, - Marker::FixExt4 => 4, - Marker::FixExt8 => 8, - Marker::FixExt16 => 16, + // fixext N = marker + 1 type byte + N data bytes + Marker::FixExt1 => 2, + Marker::FixExt2 => 3, + Marker::FixExt4 => 5, + Marker::FixExt8 => 9, + Marker::FixExt16 => 17, + // ext 8/16/32 = marker + length bytes (read here) + 1 type byte + data Marker::Ext8 => { let mut b = [0u8; 1]; rd.read_exact(&mut b).map_err(Error::from)?; let len = b[0] as usize; - 1 + len // type byte + data + 1 + len } Marker::Ext16 => { let mut b = [0u8; 2]; rd.read_exact(&mut b).map_err(Error::from)?; let len = u16::from_be_bytes(b) as usize; - 2 + len // type bytes + data + 1 + len } Marker::Ext32 => { let mut b = [0u8; 4]; rd.read_exact(&mut b).map_err(Error::from)?; let len = u32::from_be_bytes(b) as usize; - 4 + len // type bytes + data + 1 + len } Marker::Reserved => 0, }; diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index d33be0d2f..44ef9e25a 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -826,7 +826,10 @@ impl FileMetaVersionHeader { pub fn matches_not_strict(&self, o: &FileMetaVersionHeader) -> bool { let mut ok = self.version_id == o.version_id && self.version_type == o.version_type && self.matches_ec(o); - if self.version_id.is_none() { + // Disk-loaded headers keep the null version as Some(nil), not None: all null + // versions share one id, so mod_time is the only thing distinguishing an + // interrupted overwrite from the committed version. + if self.version_id.is_none() || self.version_id == Some(Uuid::nil()) { ok = ok && self.mod_time == o.mod_time; } @@ -2916,11 +2919,14 @@ pub async fn read_xl_meta_no_data(reader: &mut R, size: us return Err(Error::FileCorrupt); } - let tmp = &buf[want..]; + // The metadata block is followed by a 5-byte msgp uint32 CRC trailer; + // a file truncated inside the trailer is corrupt, not a shorter meta. let crc_size = 5; - let other_size = tmp.len() - crc_size; + if buf.len() - want < crc_size { + return Err(Error::FileCorrupt); + } - want += tmp.len() - other_size; + want += crc_size; buf.truncate(want); Ok(buf)