From daa7435bdf1bd7ed2e2cf69a52ca652fd70f8295 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 18 Aug 2026 02:48:29 +0800 Subject: [PATCH] refactor(replication): move resyncer pure decision logic into rustfs-replication (M2) Pure-move milestone M2 of the ECStore replication split (backlog#1675 P1-17): relocate the resyncer's IO-free decision helpers, with their unit tests, into the crates they already belong to by type ownership. No behavior change. Moved into crates/replication: - resync.rs: resync_status_duration - delete.rs: resync_existing_delete_replication_info, replicate_delete_outcome, target_delete_version_id, delete_marker_purge_version_id, delete_marker_purge_mrf_entry - object.rs: version_identity_drifted, is_replication_target_offline_error, SsecPassthroughCapability, SsecPassthroughGate, ssec_passthrough_gate, ssec_passthrough_evidence_present (param-demoted to the echoed customer-algorithm string; ECStore keeps the HeadObjectOutput adapter) - filemeta.rs: NULL_VERSION_ID wire literal (crate-owned copy per the filemeta-independence contract) ECStore rewiring (Rule #14: imports stay in *_boundary.rs): - resync/object-decision/target boundaries re-export the moved symbols; resyncer call sites are unchanged - bucket_target_sys keeps only the verdict cache + TTL and re-exports the capability enum so existing consumer paths keep compiling Not moved (signatures carry ECStore or aws-sdk types): verify_resync_head_result, resync_target_error_detail, the SdkError classifiers, the replicate_all_* option/info builders, and the env-coupled bounded_resync_max_jobs admission clamp. README milestone table updated. --- .../ecstore/src/bucket/bucket_target_sys.rs | 28 +- .../ecstore/src/bucket/replication/README.md | 2 +- crates/ecstore/src/bucket/replication/mod.rs | 1 + .../replication_filemeta_boundary.rs | 7 +- .../replication_object_decision_boundary.rs | 11 +- .../replication_resync_boundary.rs | 3 +- .../replication/replication_resyncer.rs | 336 +----------------- .../replication_target_boundary.rs | 106 +----- crates/replication/src/delete.rs | 206 ++++++++++- crates/replication/src/filemeta.rs | 5 + crates/replication/src/lib.rs | 14 +- crates/replication/src/object.rs | 213 ++++++++++- crates/replication/src/resync.rs | 41 +++ 13 files changed, 518 insertions(+), 455 deletions(-) diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 3b354622f..fe04f0e2b 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -299,25 +299,15 @@ struct TargetClientBuildProbe { release: Arc, } -/// Whether a replication target preserves the SSE-C passthrough transport -/// headers (`X-Rustfs-Replication-*`) end to end. -/// -/// A target that silently drops those headers (MinIO, generic S3) stores the -/// forwarded ciphertext without its decryption material — an unreadable -/// replica that used to report COMPLETED. The replication worker audits the -/// first passthrough PUT per target (HEAD-back for SSE-C evidence) and caches -/// the verdict here; a fresh `Unsupported` fails SSE-C replication closed -/// before any PUT is sent. Entries follow the `arn_remotes_map` lifecycle -/// (rebuilding or removing a target resets its capability to `Unknown`) and -/// additionally expire after [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], after which -/// the next attempt re-audits. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] -pub enum SsecPassthroughCapability { - #[default] - Unknown, - Supported, - Unsupported, -} +/// SSE-C passthrough capability verdicts (see the enum's own docs in +/// `rustfs-replication`) are cached here per target ARN: entries follow the +/// `arn_remotes_map` lifecycle (rebuilding or removing a target resets its +/// capability to `Unknown`) and additionally expire after +/// [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], after which the next attempt +/// re-audits. Re-exported so existing `bucket_target_sys` consumers keep +/// their import path while the verdict vocabulary lives with the +/// replication decision logic. +pub use crate::bucket::replication::SsecPassthroughCapability; /// How long an audited SSE-C passthrough verdict stays authoritative. /// diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index 612e85e9e..cd69bb0d0 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -128,7 +128,7 @@ Target end state: |---|---|---| | M0 | Record the completion criteria and end state (this section). | Done | | M1 | Contract extraction: resync/queue/stats/object-decision/filemeta/storage wire contracts owned by `crates/replication`; ECStore imports concentrated in `*_boundary.rs`; event sink and runtime access behind local contracts. | Done — see Required Contracts | -| M2 | Move resyncer pure decision logic (no IO) into `crates/replication`. | Pending; sequence after splitting the oversized resyncer/pool functions (`resync_bucket`, `replicate_all`, `start_mrf_processor`) so moves stay mechanical | +| M2 | Move resyncer pure decision logic (no IO) into `crates/replication`. | Done — moved the pure decision helpers with their unit tests: `resync_status_duration` (resync), `resync_existing_delete_replication_info` / `replicate_delete_outcome` / `target_delete_version_id` / `delete_marker_purge_version_id` / `delete_marker_purge_mrf_entry` (delete), `version_identity_drifted` / `is_replication_target_offline_error` / the SSE-C passthrough gate family incl. `SsecPassthroughCapability` (object; `ssec_passthrough_evidence_present` was param-demoted to the echoed customer-algorithm string, ECStore keeps the `HeadObjectOutput` adapter). ECStore imports them through the resync/object-decision/target boundaries; `bucket_target_sys` keeps only the verdict cache + TTL and re-exports the capability enum. Not moved (signatures carry ECStore or aws-sdk types): `verify_resync_head_result`, `resync_target_error_detail`, the `SdkError` classifiers (`has_raw_status`, `is_version_id_format_mismatch`), the `replicate_all_*` option/info builders, and the env-coupled `bounded_resync_max_jobs` admission clamp. | | M3 | Move the worker runtime (`replication_pool.rs`, the IO paths of `replication_resyncer.rs`, `replication_state.rs`) once the contract traits are stable. Highest-risk step of the whole plan; do it last. | Pending | | M4 | Retire the boundary modules together with their guard-script entries; delete `datatypes.rs`. | Pending | diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index d076216a3..9d23d3a7e 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -85,4 +85,5 @@ pub use replication_scanner_bridge::ReplicationScannerBridge; pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog}; pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats}; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; +pub use replication_target_boundary::SsecPassthroughCapability; pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index 5a89e0fb2..917ac65a8 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -12,12 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) use rustfs_filemeta::NULL_VERSION_ID; pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry}; pub(crate) use rustfs_replication::{ - REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, - ReplicatedTargetInfo, ReplicationAction, ReplicationWorkerOperation, ResyncDecision, get_replication_state, - parse_replicate_decision, replicate_decision_for_admitted_targets, target_reset_header, version_purge_statuses_map, + REPLICATE_EXISTING, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, + ReplicationWorkerOperation, ResyncDecision, get_replication_state, parse_replicate_decision, + replicate_decision_for_admitted_targets, target_reset_header, version_purge_statuses_map, }; pub use rustfs_replication::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType, diff --git a/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs b/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs index d6c4f1ceb..4360ed4ef 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs @@ -18,9 +18,10 @@ pub use rustfs_replication::{ should_use_existing_delete_replication_source, }; pub(crate) use rustfs_replication::{ - ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, - delete_replication_missing_source_decision, delete_replication_object_opts, heal_uses_delete_replication_path, - is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match, - replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_target_for_object, - should_retry_delete_marker_purge, + ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, delete_marker_purge_mrf_entry, + delete_marker_purge_version_id, delete_replication_missing_source_decision, delete_replication_object_opts, + heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication, + replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size, + replication_multipart_part_plan, resync_existing_delete_replication_info, resync_target_for_object, + should_retry_delete_marker_purge, target_delete_version_id, }; diff --git a/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs b/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs index 6162e1d98..32e3f6c61 100644 --- a/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs @@ -22,7 +22,8 @@ use super::replication_filemeta_boundary::MrfReplicateEntry; pub(crate) use rustfs_replication::should_count_head_proxy_failure; pub use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus}; pub(crate) use rustfs_replication::{ - is_version_id_mismatch, resync_state_accepts_update, sanitize_resync_error_detail, should_auto_resume_resync, + is_version_id_mismatch, resync_state_accepts_update, resync_status_duration, sanitize_resync_error_detail, + should_auto_resume_resync, }; #[allow( diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 57bae1c90..882ece91c 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -18,10 +18,9 @@ use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_not_found}; use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ - MrfReplicateEntry, NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, - ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, - ReplicationWorkerOperation, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, - replication_statuses_map, target_reset_header, version_purge_statuses_map, + REPLICATE_EXISTING, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, + ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, get_replication_state, + parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; use super::replication_lock_boundary::ReplicationLockTiming; use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC}; @@ -30,9 +29,11 @@ use super::replication_metadata_boundary::ReplicationMetadataStore; use super::replication_msgp_boundary::ReplicationMsgpCodec; use super::replication_object_config::{ReplicationConfig, get_replication_config, must_replicate}; use super::replication_object_decision_boundary::{ - MustReplicateOptions, ReplicationMultipartPartInput, heal_uses_delete_replication_path, - is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match, - replication_multipart_complete_actual_size, replication_multipart_part_plan, should_retry_delete_marker_purge, + MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, + heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication, + replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size, + replication_multipart_part_plan, resync_existing_delete_replication_info, should_retry_delete_marker_purge, + target_delete_version_id, }; use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission}; use super::replication_resync_boundary::ResyncStatusType; @@ -40,21 +41,23 @@ use super::replication_resync_boundary::ResyncStatusType; use super::replication_resync_boundary::should_count_head_proxy_failure; use super::replication_resync_boundary::{ BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, encode_resync_file, is_version_id_mismatch, - resync_state_accepts_update, sanitize_resync_error_detail, + resync_state_accepts_update, resync_status_duration, sanitize_resync_error_detail, }; #[cfg(test)] use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX, decode_resync_file}; +#[cfg(test)] +use super::replication_storage_boundary::ReplicationDeletedObject; use super::replication_storage_boundary::{ AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, - ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions, + ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions, }; use super::replication_target_boundary::{ ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, - SsecPassthroughCapability, SsecPassthroughGate, TargetClient, replication_action_for_target_head, - replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, - replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size, - replication_put_object_options, replication_target_head_is_newer_null_version, resolve_read_api_version_id, - ssec_passthrough_evidence_present, ssec_passthrough_gate, + SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error, + replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options, + replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted, + replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version, + resolve_read_api_version_id, ssec_passthrough_evidence_present, ssec_passthrough_gate, version_identity_drifted, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -110,21 +113,6 @@ const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_ const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf"; const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total"; const EVENT_REPLICATION_VERSION_IDENTITY_DRIFT: &str = "replication_version_identity_drift"; -const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[ - "dispatch failure", - "timeouterror", - "timed out", - "connection refused", - "connection reset", - "connection closed", - "connection aborted", - "broken pipe", - "dns error", - "failed to lookup address", - "name or service not known", - "deadline has elapsed", - "tcp connect error", -]; #[allow( dead_code, @@ -194,27 +182,6 @@ const METRIC_VERSION_IDENTITY_DRIFT_TOTAL: &str = "rustfs_replication_version_id /// after a restart is acceptable. static VERSION_IDENTITY_WARNED_ARNS: LazyLock>> = LazyLock::new(|| StdMutex::new(HashSet::new())); -/// Runtime half of the P1-19 version-identity contract (the explicit probe -/// lives in replication-check's VersionFidelity phase): every replication PUT -/// response reveals whether the target adopted the source version id. A -/// target minting its own ids silently breaks version-addressed deletes and -/// heal, so surface it — once per target — instead of letting the divergence -/// accumulate unseen. -/// Pure drift judgment: the contract only applies when the source addressed a -/// real (non-nil) version uuid, and drift means the target answered with -/// anything else — including nothing at all. -fn version_identity_drifted(source_version_id: &str, assigned_version_id: Option<&str>) -> bool { - if source_version_id.is_empty() { - return false; - } - // A nil source uuid travels as the literal "null" (unversioned-source - // semantics); no identity contract applies to it. - if Uuid::parse_str(source_version_id).map(|uuid| uuid.is_nil()).unwrap_or(true) { - return false; - } - assigned_version_id != Some(source_version_id) -} - fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &str, assigned_version_id: Option<&str>) { if !version_identity_drifted(source_version_id, assigned_version_id) { return; @@ -257,13 +224,6 @@ fn is_version_id_format_mismatch(err: &SdkError) -> bool { is_version_id_mismatch(code, raw_status) } -fn is_replication_target_offline_error(err: &(impl Display + ?Sized)) -> bool { - let message = err.to_string().to_ascii_lowercase(); - REPLICATION_TARGET_OFFLINE_ERROR_MARKERS - .iter() - .any(|marker| message.contains(marker)) -} - async fn mark_replication_target_offline_if_needed(target_client: &Arc, err: &(impl Display + ?Sized)) { if is_replication_target_offline_error(err) { ReplicationTargetStore::mark_target_offline(target_client).await; @@ -391,31 +351,6 @@ async fn audit_ssec_passthrough_replica( static RESYNC_WORKER_COUNT: usize = 10; -fn resync_status_duration( - status: ResyncStatusType, - start_time: Option, - now: OffsetDateTime, -) -> Option { - if !matches!( - status, - ResyncStatusType::ResyncCompleted | ResyncStatusType::ResyncFailed | ResyncStatusType::ResyncCanceled - ) { - return None; - } - - let millis = (now - start_time?).whole_milliseconds(); - if millis < 0 { - return None; - } - - let millis = if millis > i128::from(u64::MAX) { - u64::MAX - } else { - u64::try_from(millis).ok()? - }; - Some(std::time::Duration::from_millis(millis)) -} - type ResyncCancelKey = (String, String, String); fn configured_resync_max_jobs() -> usize { @@ -1213,33 +1148,6 @@ fn spawn_resync_walk_task( (rx, walk_failed, walk_task) } -/// Build the delete-replication work item for an existing delete marker or -/// version purge discovered during a resync scan. -fn resync_existing_delete_replication_info(roi: &ReplicateObjectInfo, target_arn: &str) -> DeletedObjectReplicationInfo { - let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() { - (None, roi.version_id) - } else { - (roi.version_id, None) - }; - - DeletedObjectReplicationInfo { - delete_object: ReplicationDeletedObject { - object_name: roi.name.clone(), - delete_marker_version_id: dm_version_id, - version_id, - replication_state: roi.replication_state.clone(), - delete_marker: roi.delete_marker, - delete_marker_mtime: roi.mod_time, - ..Default::default() - }, - bucket: roi.bucket.clone(), - event_type: REPLICATE_EXISTING_DELETE.to_string(), - op_type: ReplicationType::ExistingObject, - target_arn: target_arn.to_string(), - ..Default::default() - } -} - /// Classify the target HEAD verification result for one resynced object, /// updating the per-object status counters and returning the accounted size /// together with any verification error. @@ -1956,29 +1864,6 @@ pub(crate) async fn replicate_delete_with_outcome( ) } -/// Whether a delete replication fully succeeded — the MRF replay acknowledges -/// (drops) an entry exactly when this returns true. -/// -/// The delayed purge is deliberately NOT an input: holding the outcome hostage -/// to it (`&& !requires_delayed_purge`) forced `false` for every delete-marker -/// entry and retained them all in the durable MRF journal forever. Purge -/// failures persist their own purge-intent entry instead -/// (`watch_and_purge_source_delete_marker`), and replays of those entries -/// report purge success through `purge_stale_delete_marker_targets`. -fn replicate_delete_outcome( - expected_targets: usize, - replicated_targets: usize, - state_persisted: bool, - source_state_verified: bool, - replication_status: &ReplicationStatusType, -) -> bool { - expected_targets > 0 - && replicated_targets == expected_targets - && state_persisted - && source_state_verified - && *replication_status == ReplicationStatusType::Completed -} - async fn source_delete_marker_missing( storage: &S, bucket: &str, @@ -2003,29 +1888,6 @@ async fn source_delete_marker_missing( } } -/// Which version a delete-marker purge should address on one target. -/// -/// `None` means do not purge at all: the recorded mapping disagreed across the -/// dual internal prefixes, and guessing an id could destroy a live version on -/// the target. `Some(id)` is the exact version the target reported when it -/// accepted the marker; falling back to a source-derived id is only correct -/// when the target mirrors source version ids, which a generic S3 target does -/// not. -fn delete_marker_purge_version_id( - state: Option<&ReplicationState>, - arn: &str, - delete_marker_version_id: Uuid, -) -> Option> { - if state.is_some_and(|state| state.target_delete_marker_version_ids_corrupt) { - return None; - } - let recorded = state.and_then(|state| state.target_delete_marker_version_ids.get(arn).cloned()); - Some(match recorded { - Some(version_id) => Some(version_id), - None => target_delete_version_id(delete_marker_version_id, true), - }) -} - /// One purge pass over the eligible targets. Returns the ARNs that must be /// retried: the remote DELETE failed, or the target client was unavailable /// (e.g. a runtime cache miss). Inconsistent recorded version mappings are a @@ -2193,20 +2055,6 @@ async fn watch_and_purge_source_delete_marker( } } -/// Shape an exhausted purge intent as a marker-creation delete entry. Replay -/// reconstructs it with `delete_marker: true`, finds the source marker gone, -/// and funnels into the stale-marker branch of `replicate_delete_with_outcome` -/// — which re-runs the purge without touching source state and reports purge -/// success as the replay outcome. -fn delete_marker_purge_mrf_entry(dobj: &DeletedObjectReplicationInfo, failed_arns: Vec) -> MrfReplicateEntry { - let mut entry = dobj.to_mrf_entry(); - entry.delete_marker = true; - entry.version_id = None; - entry.retry_count = 0; - entry.target_arns = failed_arns; - entry -} - async fn enqueue_delete_marker_purge_mrf(dobj: &DeletedObjectReplicationInfo, failed_arns: Vec) { let arns = failed_arns.join(","); let miss_reason = match runtime_sources::replication_pool() { @@ -2566,14 +2414,6 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted all_succeeded } -fn target_delete_version_id(version_id: Uuid, version_purge: bool) -> Option { - if version_id.is_nil() { - version_purge.then(|| NULL_VERSION_ID.to_string()) - } else { - Some(version_id.to_string()) - } -} - async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc) -> ReplicatedTargetInfo { let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id { version_id.to_owned() @@ -4085,27 +3925,6 @@ mod tests { ReplicationTargetStore::register_test_target(target).await; } - /// P1-19 runtime spot-check exemption matrix: drift only applies when the - /// source addressed a real version uuid. - #[test] - fn test_version_identity_drift_judgment() { - let source = "6fa459ea-ee8a-3ca4-894e-db77e160355e"; - for (sent, got, expected) in [ - (source, Some(source), false), - (source, Some("0e304ce5-33e9-4b8a-9b12-9e40a53e6ded"), true), - (source, None, true), - ("", None, false), - ("null", Some("anything"), false), - ("00000000-0000-0000-0000-000000000000", Some("anything"), false), - ] { - assert_eq!( - version_identity_drifted(sent, got), - expected, - "sent {sent:?} got {got:?} must judge drift = {expected}" - ); - } - } - #[test] fn resync_admission_configuration_is_bounded() { assert_eq!(ENV_REPL_RESYNC_MAX_JOBS, "RUSTFS_REPL_RESYNC_MAX_JOBS"); @@ -4148,15 +3967,6 @@ mod tests { drop((first, second)); } - #[test] - fn replication_target_offline_error_classifier_is_network_scoped() { - assert!(is_replication_target_offline_error(&"put_object dispatch failure: connector error")); - assert!(is_replication_target_offline_error(&"request TimeoutError after retry")); - assert!(is_replication_target_offline_error(&"tcp connect error: connection refused")); - assert!(!is_replication_target_offline_error(&"put_object failed: AccessDenied: denied")); - assert!(!is_replication_target_offline_error(&"put_object failed: NoSuchBucket")); - } - #[tokio::test] async fn replication_target_network_failure_marks_target_offline() { let endpoint = format!("http://network-failure-{}.example:9000", Uuid::new_v4()); @@ -4430,27 +4240,6 @@ mod tests { ); } - /// P1-21 regression guard for the outcome formula. A fully successful - /// delete-marker replication must acknowledge its MRF entry: the formula - /// once carried `&& !requires_delayed_purge`, which pinned every - /// delete-marker entry to Missed and retained the whole backlog forever. - /// (Deterministically staging a marker-creation entry in the durable - /// journal from e2e would require saturating the worker queues, so the - /// formula is pinned here instead; the purge-intent replay half is pinned - /// by the delayed-purge e2e pair.) - #[test] - fn test_replicate_delete_outcome_is_not_held_hostage_by_the_delayed_purge() { - assert!( - replicate_delete_outcome(1, 1, true, true, &ReplicationStatusType::Completed), - "a completed delete-marker replication must be acknowledgeable even though a delayed purge watch is pending" - ); - assert!(!replicate_delete_outcome(0, 0, true, true, &ReplicationStatusType::Completed)); - assert!(!replicate_delete_outcome(2, 1, true, true, &ReplicationStatusType::Completed)); - assert!(!replicate_delete_outcome(1, 1, false, true, &ReplicationStatusType::Completed)); - assert!(!replicate_delete_outcome(1, 1, true, false, &ReplicationStatusType::Completed)); - assert!(!replicate_delete_outcome(1, 1, true, true, &ReplicationStatusType::Failed)); - } - /// P1-21 review follow-up: a target whose recorded marker version is /// inconsistent must be reported as a per-target FAILURE. Treating the /// refusal as success let the watcher and the MRF replay drop the purge @@ -4490,41 +4279,6 @@ mod tests { ); } - #[test] - fn test_delete_marker_purge_mrf_entry_replays_through_the_stale_marker_branch() { - let delete_marker_version_id = Uuid::new_v4(); - let dobj = DeletedObjectReplicationInfo { - delete_object: ReplicationDeletedObject { - object_name: "doc.txt".to_string(), - // A version-purge flavored source event: the entry must still - // be reshaped as a marker-creation delete so replay funnels - // into the stale-marker branch instead of re-running the full - // delete replication (whose source-state stamping would fail - // against the already-purged version). - delete_marker: false, - version_id: Some(Uuid::new_v4()), - delete_marker_version_id: Some(delete_marker_version_id), - ..Default::default() - }, - bucket: "bucket-a".to_string(), - ..Default::default() - }; - - let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]); - - assert!(entry.delete_marker, "purge intents must replay as marker-creation deletes"); - assert_eq!(entry.version_id, None, "the purged data version must not leak into the replay"); - assert_eq!(entry.delete_marker_version_id, Some(delete_marker_version_id)); - assert_eq!( - entry.target_arns, - vec!["arn:a".to_string()], - "only the targets whose purge failed may be retried" - ); - assert_eq!(entry.retry_count, 0); - assert_eq!(entry.bucket, "bucket-a"); - assert_eq!(entry.object, "doc.txt"); - } - #[test] fn test_is_retryable_delete_replication_head_error_allows_delete_marker_head_responses() { assert!( @@ -5032,60 +4786,4 @@ mod tests { assert!(resync_state_accepts_update(¤t, &matching)); assert!(!resync_state_accepts_update(¤t, &stale)); } - - #[test] - fn test_resync_status_duration_only_tracks_terminal_status() { - let start = match OffsetDateTime::from_unix_timestamp(1_700_000_000) { - Ok(start) => start, - Err(err) => panic!("valid test timestamp: {err}"), - }; - let end = start + time::Duration::seconds(2); - - assert_eq!( - resync_status_duration(ResyncStatusType::ResyncCompleted, Some(start), end), - Some(std::time::Duration::from_millis(2000)) - ); - assert_eq!(resync_status_duration(ResyncStatusType::ResyncStarted, Some(start), end), None); - assert_eq!(resync_status_duration(ResyncStatusType::ResyncFailed, None, end), None); - } - - #[test] - fn target_delete_version_id_preserves_explicit_null_purges() { - let version_id = Uuid::new_v4(); - - assert_eq!(target_delete_version_id(version_id, true), Some(version_id.to_string())); - assert_eq!(target_delete_version_id(Uuid::nil(), true).as_deref(), Some(NULL_VERSION_ID)); - assert_eq!(target_delete_version_id(Uuid::nil(), false), None); - } - - #[test] - fn delete_marker_purge_prefers_the_recorded_target_version() { - let source = Uuid::new_v4(); - let arn = "arn:rustfs:replication::target:bucket"; - - // No recorded mapping: fall back to deriving from the source uuid. - assert_eq!(delete_marker_purge_version_id(None, arn, source), Some(Some(source.to_string()))); - - // Recorded mapping wins — a generic S3 target assigns its own id, so the - // derived one would purge the wrong version or nothing at all. - let mut state = ReplicationState::default(); - state - .target_delete_marker_version_ids - .insert(arn.to_string(), "target-assigned-id".to_string()); - assert_eq!( - delete_marker_purge_version_id(Some(&state), arn, source), - Some(Some("target-assigned-id".to_string())) - ); - - // A mapping recorded for a different ARN must not be reused. - assert_eq!( - delete_marker_purge_version_id(Some(&state), "arn:rustfs:replication::other:bucket", source), - Some(Some(source.to_string())) - ); - - // Inconsistent persisted metadata: refuse to purge rather than guess. - let mut corrupt = state.clone(); - corrupt.target_delete_marker_version_ids_corrupt = true; - assert_eq!(delete_marker_purge_version_id(Some(&corrupt), arn, source), None); - } } diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index d060116d5..87d94779e 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -36,12 +36,15 @@ use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; pub(crate) use crate::bucket::bucket_target_sys::{ - AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, SsecPassthroughCapability, TargetClient, - resolve_read_api_version_id, + AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, resolve_read_api_version_id, }; #[cfg(test)] pub(crate) use crate::bucket::target::BucketTarget; pub(crate) use crate::bucket::target::BucketTargets; +pub use rustfs_replication::SsecPassthroughCapability; +pub(crate) use rustfs_replication::{ + SsecPassthroughGate, is_replication_target_offline_error, ssec_passthrough_gate, version_identity_drifted, +}; use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result}; @@ -149,52 +152,11 @@ pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap SsecPassthroughGate { - if !ssec { - return SsecPassthroughGate::Proceed; - } - // An expired verdict — Supported or Unsupported — must be re-earned: a - // stale Unsupported would otherwise stick forever after a target upgrade, - // and a stale Supported would fail open after a backend swap behind the - // same endpoint. - if expired { - return SsecPassthroughGate::ProceedWithAudit; - } - match capability { - SsecPassthroughCapability::Supported => SsecPassthroughGate::Proceed, - SsecPassthroughCapability::Unknown => SsecPassthroughGate::ProceedWithAudit, - SsecPassthroughCapability::Unsupported => SsecPassthroughGate::FailClosed, - } -} - -/// True when a replication-check HEAD of the replica proves the SSE-C -/// material survived passthrough: a RustFS target restores the transport -/// headers into the stored SSE-C keys and its HEAD echoes -/// `x-amz-server-side-encryption-customer-algorithm` (the replication-check -/// exemption skips key validation but not the metadata echo). A target that -/// dropped the headers stored a plain object and echoes nothing. +/// HeadObjectOutput adapter over the pure SSE-C passthrough evidence +/// judgment owned by `rustfs-replication`: extract the echoed +/// customer-algorithm header and let the crate-owned policy decide. pub(crate) fn ssec_passthrough_evidence_present(head: &HeadObjectOutput) -> bool { - head.sse_customer_algorithm.as_deref().is_some_and(|algo| !algo.is_empty()) + rustfs_replication::ssec_passthrough_evidence_present(head.sse_customer_algorithm.as_deref()) } pub(crate) struct ReplicationTargetStore; @@ -960,53 +922,9 @@ mod tests { } } - /// N2 fail-closed policy: SSE-C replication may only proceed silently - /// against a target with a FRESH proof that it preserves the passthrough - /// transport headers. Unknown targets must be audited; freshly-flagged - /// dropping targets must never receive the PUT; an expired verdict in - /// EITHER direction must be re-earned through the audit — a sticky - /// Unsupported would outlive a target upgrade, and a sticky Supported - /// would fail open after a backend swap behind the same endpoint. - #[test] - fn ssec_passthrough_gate_is_fail_closed_and_ttl_bounded() { - for capability in [ - SsecPassthroughCapability::Unknown, - SsecPassthroughCapability::Supported, - SsecPassthroughCapability::Unsupported, - ] { - for expired in [false, true] { - assert_eq!( - ssec_passthrough_gate(false, capability, expired), - SsecPassthroughGate::Proceed, - "non-SSE-C objects must never be gated on the passthrough capability" - ); - } - } - assert_eq!( - ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, false), - SsecPassthroughGate::Proceed - ); - assert_eq!( - ssec_passthrough_gate(true, SsecPassthroughCapability::Unknown, false), - SsecPassthroughGate::ProceedWithAudit - ); - assert_eq!( - ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, false), - SsecPassthroughGate::FailClosed - ); - // Expiry flips both directions back to the audit. - assert_eq!( - ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, true), - SsecPassthroughGate::ProceedWithAudit, - "an expired Unsupported verdict must allow a re-audit (upgraded target recovers without operator action)" - ); - assert_eq!( - ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, true), - SsecPassthroughGate::ProceedWithAudit, - "an expired Supported verdict must be re-proven (backend swap behind the same endpoint must not fail open)" - ); - } - + /// Pins the HeadObjectOutput field extraction feeding the crate-owned + /// evidence judgment (the gate/evidence policy matrix itself is pinned in + /// `rustfs-replication`'s object tests). #[test] fn ssec_passthrough_evidence_requires_customer_algorithm_echo() { let with_evidence = HeadObjectOutput::builder().sse_customer_algorithm("AES256").build(); diff --git a/crates/replication/src/delete.rs b/crates/replication/src/delete.rs index 3a0c32bf8..3b6fbb016 100644 --- a/crates/replication/src/delete.rs +++ b/crates/replication/src/delete.rs @@ -14,8 +14,13 @@ use std::any::Any; +use uuid::Uuid; + use crate::storage_api::DeletedObject; -use crate::{MrfOpKind, MrfReplicateEntry, ReplicationState, ReplicationType, ReplicationWorkerOperation}; +use crate::{ + MrfOpKind, MrfReplicateEntry, NULL_VERSION_ID, REPLICATE_EXISTING_DELETE, ReplicateObjectInfo, ReplicationState, + ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, +}; #[derive(Debug, Clone, Default)] pub struct DeletedObjectReplicationInfo { @@ -117,17 +122,114 @@ pub fn is_retryable_delete_replication_head_error(is_not_found: bool, code: Opti !(is_not_found || matches!(code, Some("MethodNotAllowed" | "405"))) } +/// Build the delete-replication work item for an existing delete marker or +/// version purge discovered during a resync scan. +pub fn resync_existing_delete_replication_info(roi: &ReplicateObjectInfo, target_arn: &str) -> DeletedObjectReplicationInfo { + let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() { + (None, roi.version_id) + } else { + (roi.version_id, None) + }; + + DeletedObjectReplicationInfo { + delete_object: DeletedObject { + object_name: roi.name.clone(), + delete_marker_version_id: dm_version_id, + version_id, + replication_state: roi.replication_state.clone(), + delete_marker: roi.delete_marker, + delete_marker_mtime: roi.mod_time, + ..Default::default() + }, + bucket: roi.bucket.clone(), + event_type: REPLICATE_EXISTING_DELETE.to_string(), + op_type: ReplicationType::ExistingObject, + target_arn: target_arn.to_string(), + ..Default::default() + } +} + +/// Whether a delete replication fully succeeded — the MRF replay acknowledges +/// (drops) an entry exactly when this returns true. +/// +/// The delayed purge is deliberately NOT an input: holding the outcome hostage +/// to it (`&& !requires_delayed_purge`) forced `false` for every delete-marker +/// entry and retained them all in the durable MRF journal forever. Purge +/// failures persist their own purge-intent entry instead +/// (`watch_and_purge_source_delete_marker`), and replays of those entries +/// report purge success through `purge_stale_delete_marker_targets`. +pub fn replicate_delete_outcome( + expected_targets: usize, + replicated_targets: usize, + state_persisted: bool, + source_state_verified: bool, + replication_status: &ReplicationStatusType, +) -> bool { + expected_targets > 0 + && replicated_targets == expected_targets + && state_persisted + && source_state_verified + && *replication_status == ReplicationStatusType::Completed +} + +pub fn target_delete_version_id(version_id: Uuid, version_purge: bool) -> Option { + if version_id.is_nil() { + version_purge.then(|| NULL_VERSION_ID.to_string()) + } else { + Some(version_id.to_string()) + } +} + +/// Which version a delete-marker purge should address on one target. +/// +/// `None` means do not purge at all: the recorded mapping disagreed across the +/// dual internal prefixes, and guessing an id could destroy a live version on +/// the target. `Some(id)` is the exact version the target reported when it +/// accepted the marker; falling back to a source-derived id is only correct +/// when the target mirrors source version ids, which a generic S3 target does +/// not. +pub fn delete_marker_purge_version_id( + state: Option<&ReplicationState>, + arn: &str, + delete_marker_version_id: Uuid, +) -> Option> { + if state.is_some_and(|state| state.target_delete_marker_version_ids_corrupt) { + return None; + } + let recorded = state.and_then(|state| state.target_delete_marker_version_ids.get(arn).cloned()); + Some(match recorded { + Some(version_id) => Some(version_id), + None => target_delete_version_id(delete_marker_version_id, true), + }) +} + +/// Shape an exhausted purge intent as a marker-creation delete entry. Replay +/// reconstructs it with `delete_marker: true`, finds the source marker gone, +/// and funnels into the stale-marker branch of `replicate_delete_with_outcome` +/// — which re-runs the purge without touching source state and reports purge +/// success as the replay outcome. +pub fn delete_marker_purge_mrf_entry(dobj: &DeletedObjectReplicationInfo, failed_arns: Vec) -> MrfReplicateEntry { + let mut entry = dobj.to_mrf_entry(); + entry.delete_marker = true; + entry.version_id = None; + entry.retry_count = 0; + entry.target_arns = failed_arns; + entry +} + #[cfg(test)] mod tests { use std::collections::HashMap; use super::{ - DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication, - should_retry_delete_marker_purge, + DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, + is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, + should_retry_delete_marker_purge, target_delete_version_id, }; use crate::storage_api::DeletedObject; use crate::{ - MrfOpKind, ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, VersionPurgeStatusType, + MrfOpKind, NULL_VERSION_ID, ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, + VersionPurgeStatusType, }; use uuid::Uuid; @@ -328,4 +430,100 @@ mod tests { assert!(!is_retryable_delete_replication_head_error(true, Some("NoSuchKey"))); assert!(is_retryable_delete_replication_head_error(false, Some("AccessDenied"))); } + + /// P1-21 regression guard for the outcome formula. A fully successful + /// delete-marker replication must acknowledge its MRF entry: the formula + /// once carried `&& !requires_delayed_purge`, which pinned every + /// delete-marker entry to Missed and retained the whole backlog forever. + /// (Deterministically staging a marker-creation entry in the durable + /// journal from e2e would require saturating the worker queues, so the + /// formula is pinned here instead; the purge-intent replay half is pinned + /// by the delayed-purge e2e pair.) + #[test] + fn test_replicate_delete_outcome_is_not_held_hostage_by_the_delayed_purge() { + assert!( + replicate_delete_outcome(1, 1, true, true, &ReplicationStatusType::Completed), + "a completed delete-marker replication must be acknowledgeable even though a delayed purge watch is pending" + ); + assert!(!replicate_delete_outcome(0, 0, true, true, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(2, 1, true, true, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(1, 1, false, true, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(1, 1, true, false, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(1, 1, true, true, &ReplicationStatusType::Failed)); + } + + #[test] + fn test_delete_marker_purge_mrf_entry_replays_through_the_stale_marker_branch() { + let delete_marker_version_id = Uuid::new_v4(); + let dobj = DeletedObjectReplicationInfo { + delete_object: DeletedObject { + object_name: "doc.txt".to_string(), + // A version-purge flavored source event: the entry must still + // be reshaped as a marker-creation delete so replay funnels + // into the stale-marker branch instead of re-running the full + // delete replication (whose source-state stamping would fail + // against the already-purged version). + delete_marker: false, + version_id: Some(Uuid::new_v4()), + delete_marker_version_id: Some(delete_marker_version_id), + ..Default::default() + }, + bucket: "bucket-a".to_string(), + ..Default::default() + }; + + let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]); + + assert!(entry.delete_marker, "purge intents must replay as marker-creation deletes"); + assert_eq!(entry.version_id, None, "the purged data version must not leak into the replay"); + assert_eq!(entry.delete_marker_version_id, Some(delete_marker_version_id)); + assert_eq!( + entry.target_arns, + vec!["arn:a".to_string()], + "only the targets whose purge failed may be retried" + ); + assert_eq!(entry.retry_count, 0); + assert_eq!(entry.bucket, "bucket-a"); + assert_eq!(entry.object, "doc.txt"); + } + + #[test] + fn target_delete_version_id_preserves_explicit_null_purges() { + let version_id = Uuid::new_v4(); + + assert_eq!(target_delete_version_id(version_id, true), Some(version_id.to_string())); + assert_eq!(target_delete_version_id(Uuid::nil(), true).as_deref(), Some(NULL_VERSION_ID)); + assert_eq!(target_delete_version_id(Uuid::nil(), false), None); + } + + #[test] + fn delete_marker_purge_prefers_the_recorded_target_version() { + let source = Uuid::new_v4(); + let arn = "arn:rustfs:replication::target:bucket"; + + // No recorded mapping: fall back to deriving from the source uuid. + assert_eq!(delete_marker_purge_version_id(None, arn, source), Some(Some(source.to_string()))); + + // Recorded mapping wins — a generic S3 target assigns its own id, so the + // derived one would purge the wrong version or nothing at all. + let mut state = ReplicationState::default(); + state + .target_delete_marker_version_ids + .insert(arn.to_string(), "target-assigned-id".to_string()); + assert_eq!( + delete_marker_purge_version_id(Some(&state), arn, source), + Some(Some("target-assigned-id".to_string())) + ); + + // A mapping recorded for a different ARN must not be reused. + assert_eq!( + delete_marker_purge_version_id(Some(&state), "arn:rustfs:replication::other:bucket", source), + Some(Some(source.to_string())) + ); + + // Inconsistent persisted metadata: refuse to purge rather than guess. + let mut corrupt = state.clone(); + corrupt.target_delete_marker_version_ids_corrupt = true; + assert_eq!(delete_marker_purge_version_id(Some(&corrupt), arn, source), None); + } } diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index 68064dc19..8dfdb26c5 100644 --- a/crates/replication/src/filemeta.rs +++ b/crates/replication/src/filemeta.rs @@ -27,6 +27,11 @@ use uuid::Uuid; pub const REPLICATION_RESET: &str = "replication-reset"; pub const REPLICATION_STATUS: &str = "replication-status"; +/// The S3 wire spelling of the unversioned ("null") version id. Owned here as +/// part of the replication wire contracts; `rustfs-filemeta` keeps its own +/// copy of the same literal (the crates are intentionally independent). +pub const NULL_VERSION_ID: &str = "null"; + // ReplicateQueued - replication being queued trail pub const REPLICATE_QUEUED: &str = "replicate:queue"; diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index bb0c08385..cb0761b5c 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -37,11 +37,12 @@ pub use config::{ validate_replication_config_target_arns, }; pub use delete::{ - DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication, - should_retry_delete_marker_purge, + DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, + is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, + resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id, }; pub use filemeta::{ - REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING, + NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING, REPLICATE_INCOMING_DELETE, REPLICATE_MRF, REPLICATE_QUEUED, REPLICATION_RESET, REPLICATION_STATUS, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, @@ -58,8 +59,9 @@ pub use multipart::{ replication_multipart_complete_actual_size, replication_multipart_part_plan, }; pub use object::{ - ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, - replication_etags_match, target_is_newer_than_source_null_version, + ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, content_matches_by_etag, + is_replication_target_offline_error, replication_action_for_target, replication_etags_match, + ssec_passthrough_evidence_present, ssec_passthrough_gate, target_is_newer_than_source_null_version, version_identity_drifted, }; pub use operation::{ MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteSource, ReplicationDeleteStateSource, @@ -76,7 +78,7 @@ pub use queue::{ pub use resync::{ BucketReplicationResyncStatus, Error, RESYNC_FILE_MAX_BYTES, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, decode_resync_file, encode_resync_file, is_version_id_mismatch, resync_state_accepts_update, - sanitize_resync_error_detail, should_auto_resume_resync, should_count_head_proxy_failure, + resync_status_duration, sanitize_resync_error_detail, should_auto_resume_resync, should_count_head_proxy_failure, }; pub use rule::ReplicationRuleExt; pub use runtime::{ diff --git a/crates/replication/src/object.rs b/crates/replication/src/object.rs index afbb2ef46..7ce7761ca 100644 --- a/crates/replication/src/object.rs +++ b/crates/replication/src/object.rs @@ -157,11 +157,130 @@ fn comparable_metadata(metadata: Option<&HashMap>) -> HashMap) -> bool { + if source_version_id.is_empty() { + return false; + } + // A nil source uuid travels as the literal "null" (unversioned-source + // semantics); no identity contract applies to it. + if uuid::Uuid::parse_str(source_version_id) + .map(|uuid| uuid.is_nil()) + .unwrap_or(true) + { + return false; + } + assigned_version_id != Some(source_version_id) +} + +const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[ + "dispatch failure", + "timeouterror", + "timed out", + "connection refused", + "connection reset", + "connection closed", + "connection aborted", + "broken pipe", + "dns error", + "failed to lookup address", + "name or service not known", + "deadline has elapsed", + "tcp connect error", +]; + +/// True when a target operation error reads as a network/transport failure — +/// the only class of error that should mark a replication target offline. +pub fn is_replication_target_offline_error(err: &(impl std::fmt::Display + ?Sized)) -> bool { + let message = err.to_string().to_ascii_lowercase(); + REPLICATION_TARGET_OFFLINE_ERROR_MARKERS + .iter() + .any(|marker| message.contains(marker)) +} + +/// Whether a replication target preserves the SSE-C passthrough transport +/// headers (`X-Rustfs-Replication-*`) end to end. +/// +/// A target that silently drops those headers (MinIO, generic S3) stores the +/// forwarded ciphertext without its decryption material — an unreadable +/// replica that used to report COMPLETED. The replication worker audits the +/// first passthrough PUT per target (HEAD-back for SSE-C evidence) and caches +/// the verdict; a fresh `Unsupported` fails SSE-C replication closed before +/// any PUT is sent. The verdict cache (per-ARN map, lifecycle, and TTL) is +/// owned by the runtime's bucket target system; this crate owns only the +/// verdict vocabulary and the gate policy below. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum SsecPassthroughCapability { + #[default] + Unknown, + Supported, + Unsupported, +} + +/// Fail-closed decision for an SSE-C passthrough replication attempt, derived +/// from the target's cached [`SsecPassthroughCapability`]. Pure so the policy +/// can migrate with the worker (M2) without dragging the cache along; the +/// caller computes `expired` from the cache record's age (see the runtime's +/// `SSEC_PASSTHROUGH_CAPABILITY_TTL`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SsecPassthroughGate { + /// Not an SSE-C object, or the target has a fresh proof that it preserves + /// the passthrough transport headers: replicate without a HEAD-back audit. + Proceed, + /// No usable verdict — first SSE-C attempt since the target was (re)built, + /// or the recorded verdict (in either direction) aged out: PUT, then HEAD + /// the replica back and require SSE-C evidence before reporting COMPLETED. + ProceedWithAudit, + /// The target was recently proven to drop the passthrough headers: do not + /// send the PUT, report FAILED (the object stays on the normal MRF retry + /// channel and re-audits once the verdict expires). + FailClosed, +} + +pub fn ssec_passthrough_gate(ssec: bool, capability: SsecPassthroughCapability, expired: bool) -> SsecPassthroughGate { + if !ssec { + return SsecPassthroughGate::Proceed; + } + // An expired verdict — Supported or Unsupported — must be re-earned: a + // stale Unsupported would otherwise stick forever after a target upgrade, + // and a stale Supported would fail open after a backend swap behind the + // same endpoint. + if expired { + return SsecPassthroughGate::ProceedWithAudit; + } + match capability { + SsecPassthroughCapability::Supported => SsecPassthroughGate::Proceed, + SsecPassthroughCapability::Unknown => SsecPassthroughGate::ProceedWithAudit, + SsecPassthroughCapability::Unsupported => SsecPassthroughGate::FailClosed, + } +} + +/// True when a replication-check HEAD of the replica proves the SSE-C +/// material survived passthrough: a RustFS target restores the transport +/// headers into the stored SSE-C keys and its HEAD echoes +/// `x-amz-server-side-encryption-customer-algorithm` (the replication-check +/// exemption skips key validation but not the metadata echo). A target that +/// dropped the headers stored a plain object and echoes nothing. The caller +/// extracts the echoed customer-algorithm value from its HEAD response type. +pub fn ssec_passthrough_evidence_present(sse_customer_algorithm: Option<&str>) -> bool { + sse_customer_algorithm.is_some_and(|algo| !algo.is_empty()) +} + #[cfg(test)] mod tests { use super::{ - ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, - replication_etags_match, target_is_newer_than_source_null_version, + ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, + content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target, replication_etags_match, + ssec_passthrough_evidence_present, ssec_passthrough_gate, target_is_newer_than_source_null_version, + version_identity_drifted, }; use crate::filemeta::{ReplicationAction, ReplicationType}; use crate::http::AMZ_OBJECT_LOCK_MODE; @@ -269,6 +388,96 @@ mod tests { ); } + /// P1-19 runtime spot-check exemption matrix: drift only applies when the + /// source addressed a real version uuid. + #[test] + fn test_version_identity_drift_judgment() { + let source = "6fa459ea-ee8a-3ca4-894e-db77e160355e"; + for (sent, got, expected) in [ + (source, Some(source), false), + (source, Some("0e304ce5-33e9-4b8a-9b12-9e40a53e6ded"), true), + (source, None, true), + ("", None, false), + ("null", Some("anything"), false), + ("00000000-0000-0000-0000-000000000000", Some("anything"), false), + ] { + assert_eq!( + version_identity_drifted(sent, got), + expected, + "sent {sent:?} got {got:?} must judge drift = {expected}" + ); + } + } + + #[test] + fn replication_target_offline_error_classifier_is_network_scoped() { + assert!(is_replication_target_offline_error("put_object dispatch failure: connector error")); + assert!(is_replication_target_offline_error("request TimeoutError after retry")); + assert!(is_replication_target_offline_error("tcp connect error: connection refused")); + assert!(!is_replication_target_offline_error("put_object failed: AccessDenied: denied")); + assert!(!is_replication_target_offline_error("put_object failed: NoSuchBucket")); + } + + /// N2 fail-closed policy: SSE-C replication may only proceed silently + /// against a target with a FRESH proof that it preserves the passthrough + /// transport headers. Unknown targets must be audited; freshly-flagged + /// dropping targets must never receive the PUT; an expired verdict in + /// EITHER direction must be re-earned through the audit — a sticky + /// Unsupported would outlive a target upgrade, and a sticky Supported + /// would fail open after a backend swap behind the same endpoint. + #[test] + fn ssec_passthrough_gate_is_fail_closed_and_ttl_bounded() { + for capability in [ + SsecPassthroughCapability::Unknown, + SsecPassthroughCapability::Supported, + SsecPassthroughCapability::Unsupported, + ] { + for expired in [false, true] { + assert_eq!( + ssec_passthrough_gate(false, capability, expired), + SsecPassthroughGate::Proceed, + "non-SSE-C objects must never be gated on the passthrough capability" + ); + } + } + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, false), + SsecPassthroughGate::Proceed + ); + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Unknown, false), + SsecPassthroughGate::ProceedWithAudit + ); + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, false), + SsecPassthroughGate::FailClosed + ); + // Expiry flips both directions back to the audit. + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, true), + SsecPassthroughGate::ProceedWithAudit, + "an expired Unsupported verdict must allow a re-audit (upgraded target recovers without operator action)" + ); + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, true), + SsecPassthroughGate::ProceedWithAudit, + "an expired Supported verdict must be re-proven (backend swap behind the same endpoint must not fail open)" + ); + } + + #[test] + fn ssec_passthrough_evidence_requires_customer_algorithm_echo() { + assert!(ssec_passthrough_evidence_present(Some("AES256"))); + assert!( + !ssec_passthrough_evidence_present(Some("")), + "an empty echo is not evidence of preserved SSE-C material" + ); + assert!( + !ssec_passthrough_evidence_present(None), + "a plain HEAD response must classify the target as having dropped the material" + ); + } + #[test] fn replication_action_detects_tags_and_object_lock_metadata_differences() { let mut source_metadata = HashMap::new(); diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs index 2380d9e18..144181dd2 100644 --- a/crates/replication/src/resync.rs +++ b/crates/replication/src/resync.rs @@ -309,6 +309,31 @@ pub fn is_version_id_mismatch(code: Option<&str>, raw_status: Option) -> bo } } +pub fn resync_status_duration( + status: ResyncStatusType, + start_time: Option, + now: OffsetDateTime, +) -> Option { + if !matches!( + status, + ResyncStatusType::ResyncCompleted | ResyncStatusType::ResyncFailed | ResyncStatusType::ResyncCanceled + ) { + return None; + } + + let millis = (now - start_time?).whole_milliseconds(); + if millis < 0 { + return None; + } + + let millis = if millis > i128::from(u64::MAX) { + u64::MAX + } else { + u64::try_from(millis).ok()? + }; + Some(std::time::Duration::from_millis(millis)) +} + #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct BucketReplicationResyncStatus { pub version: u16, @@ -712,6 +737,22 @@ mod tests { assert!(!should_auto_resume_resync(ResyncStatusType::ResyncFailed)); } + #[test] + fn test_resync_status_duration_only_tracks_terminal_status() { + let start = match OffsetDateTime::from_unix_timestamp(1_700_000_000) { + Ok(start) => start, + Err(err) => panic!("valid test timestamp: {err}"), + }; + let end = start + time::Duration::seconds(2); + + assert_eq!( + resync_status_duration(ResyncStatusType::ResyncCompleted, Some(start), end), + Some(std::time::Duration::from_millis(2000)) + ); + assert_eq!(resync_status_duration(ResyncStatusType::ResyncStarted, Some(start), end), None); + assert_eq!(resync_status_duration(ResyncStatusType::ResyncFailed, None, end), None); + } + #[test] fn resync_state_accepts_update_only_for_matching_run() { let current = TargetReplicationResyncStatus {