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.

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -299,25 +299,15 @@ struct TargetClientBuildProbe {
     release: Arc<tokio::sync::Semaphore>,
 }
 
-/// Whether a replication target preserves the SSE-C passthrough transport
+/// SSE-C passthrough capability verdicts (see the enum's own docs in
-/// headers (`X-Rustfs-Replication-*`) end to end.
+/// `rustfs-replication`) are cached here per target ARN: entries follow the
-///
+/// `arn_remotes_map` lifecycle (rebuilding or removing a target resets its
-/// A target that silently drops those headers (MinIO, generic S3) stores the
+/// capability to `Unknown`) and additionally expire after
-/// forwarded ciphertext without its decryption material — an unreadable
+/// [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], after which the next attempt
-/// replica that used to report COMPLETED. The replication worker audits the
+/// re-audits. Re-exported so existing `bucket_target_sys` consumers keep
-/// first passthrough PUT per target (HEAD-back for SSE-C evidence) and caches
+/// their import path while the verdict vocabulary lives with the
-/// the verdict here; a fresh `Unsupported` fails SSE-C replication closed
+/// replication decision logic.
-/// before any PUT is sent. Entries follow the `arn_remotes_map` lifecycle
+pub use crate::bucket::replication::SsecPassthroughCapability;
-/// (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,
-}
 
 /// How long an audited SSE-C passthrough verdict stays authoritative.
 ///

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 |
 

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;

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,
+    REPLICATE_EXISTING, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
-    ReplicatedTargetInfo, ReplicationAction, ReplicationWorkerOperation, ResyncDecision, get_replication_state,
+    ReplicationWorkerOperation, ResyncDecision, get_replication_state, parse_replicate_decision,
-    parse_replicate_decision, replicate_decision_for_admitted_targets, target_reset_header, version_purge_statuses_map,
+    replicate_decision_for_admitted_targets, target_reset_header, version_purge_statuses_map,
 };
 pub use rustfs_replication::{
     REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType,

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,
+    ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, delete_marker_purge_mrf_entry,
-    delete_replication_missing_source_decision, delete_replication_object_opts, heal_uses_delete_replication_path,
+    delete_marker_purge_version_id, delete_replication_missing_source_decision, delete_replication_object_opts,
-    is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match,
+    heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_target_for_object,
+    replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
-    should_retry_delete_marker_purge,
+    replication_multipart_part_plan, resync_existing_delete_replication_info, resync_target_for_object,
+    should_retry_delete_marker_purge, target_delete_version_id,
 };

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(

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,
+    REPLICATE_EXISTING, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
-    ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType,
+    ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, get_replication_state,
-    ReplicationWorkerOperation, VersionPurgeStatusType, get_replication_state, parse_replicate_decision,
+    parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
-    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,
+    MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match,
+    heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan, should_retry_delete_marker_purge,
+    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,
+    SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error,
-    replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options,
+    replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options,
-    replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size,
+    replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted,
-    replication_put_object_options, replication_target_head_is_newer_null_version, resolve_read_api_version_id,
+    replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version,
-    ssec_passthrough_evidence_present, ssec_passthrough_gate,
+    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<StdMutex<HashSet<String>>> = 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<HeadObjectError>) -> 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<TargetClient>, 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<OffsetDateTime>,
-    now: OffsetDateTime,
-) -> Option<std::time::Duration> {
-    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<S: ReplicationStorage>(
     (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<S: ReplicationStorage>(
     )
 }
 
-/// 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<S: EcstoreObjectOperations>(
     storage: &S,
     bucket: &str,
@@ -2003,29 +1888,6 @@ async fn source_delete_marker_missing<S: EcstoreObjectOperations>(
     }
 }
 
-/// 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<Option<String>> {
-    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<S: ReplicationStorage>(
     }
 }
 
-/// 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<String>) -> 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<String>) {
     let arns = failed_arns.join(",");
     let miss_reason = match runtime_sources::replication_pool() {
@@ -2566,14 +2414,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
     all_succeeded
 }
 
-fn target_delete_version_id(version_id: Uuid, version_purge: bool) -> Option<String> {
-    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<TargetClient>) -> 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(&current, &matching));
         assert!(!resync_state_accepts_update(&current, &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);
-    }
 }

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,
+    AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, resolve_read_api_version_id,
-    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<String
     rustfs_replication::is_ssec_encrypted(user_defined)
 }
 
-/// Fail-closed decision for an SSE-C passthrough replication attempt, derived
+/// HeadObjectOutput adapter over the pure SSE-C passthrough evidence
-/// from the target's cached [`SsecPassthroughCapability`]. Pure so the policy
+/// judgment owned by `rustfs-replication`: extract the echoed
-/// can migrate with the worker (M2) without dragging the cache along; the
+/// customer-algorithm header and let the crate-owned policy decide.
-/// caller computes `expired` from the cache record's age (see
-/// `SSEC_PASSTHROUGH_CAPABILITY_TTL`).
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub(crate) 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(crate) 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.
 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
+    /// Pins the HeadObjectOutput field extraction feeding the crate-owned
-    /// against a target with a FRESH proof that it preserves the passthrough
+    /// evidence judgment (the gate/evidence policy matrix itself is pinned in
-    /// transport headers. Unknown targets must be audited; freshly-flagged
+    /// `rustfs-replication`'s object tests).
-    /// 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() {
         let with_evidence = HeadObjectOutput::builder().sse_customer_algorithm("AES256").build();

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<String> {
+    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<Option<String>> {
+    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<String>) -> 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,
+        DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-        should_retry_delete_marker_purge,
+        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);
+    }
 }

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";
 

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,
+    DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    should_retry_delete_marker_purge,
+    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,
+    ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, content_matches_by_etag,
-    replication_etags_match, target_is_newer_than_source_null_version,
+    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::{

crates/replication/src/object.rs

@@ -157,11 +157,130 @@ fn comparable_metadata(metadata: Option<&HashMap<String, String>>) -> HashMap<St
     comparable
 }
 
+/// 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.
+pub 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::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,
+        ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate,
-        replication_etags_match, target_is_newer_than_source_null_version,
+        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();

crates/replication/src/resync.rs

@@ -309,6 +309,31 @@ pub fn is_version_id_mismatch(code: Option<&str>, raw_status: Option<u16>) -> bo
     }
 }
 
+pub fn resync_status_duration(
+    status: ResyncStatusType,
+    start_time: Option<OffsetDateTime>,
+    now: OffsetDateTime,
+) -> Option<std::time::Duration> {
+    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 {