fix(site-replication): harden the retry drain against review findings

Adversarial review of the drain surfaced one real race and three cheap
hardenings:

- Conditional settlement for collapsed (constant-path) iam-item /
  bucket-meta entries: the snapshot resend proves delivery of the state
  as of plan-build time, so a hook failure stamped during the delivery
  window (a newer local commit the snapshot did not contain) must
  survive the snapshot's success instead of being cleared — previously
  the peer would silently diverge until the next same-path failure.
  (The operator repair path shares this collapse but keeps its existing
  unconditional settle; the drain runs every tick and needed the guard
  first.)
- Re-check the pending_* gates on the freshly loaded state: an endpoint
  refresh can commit its pending marker mid-tick without the lifecycle
  guard.
- Do not upsert retry events for peers that already left the state —
  remove_sites pruned their entries and they can never drain again.
- Correct the mutual-exclusion comment: the repair-execution lock does
  NOT close the dry-run -> execute window (dry-run takes no lock); that
  window fails safe via the preflight token, which hashes the
  replayable entries.

New tests pin the (Some settled, None failed) generation-settle
combination the peer-edit drain depends on, and the snapshot-relative
conditional settlement.

.github/workflows/docker.yml

@@ -94,6 +94,7 @@ jobs:
       short_sha: ${{ steps.check.outputs.short_sha }}
       is_prerelease: ${{ steps.check.outputs.is_prerelease }}
       create_latest: ${{ steps.check.outputs.create_latest }}
+      source_ref: ${{ steps.check.outputs.source_ref }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
@@ -118,6 +119,7 @@ jobs:
           short_sha=""
           is_prerelease=false
           create_latest=false
+          source_ref="$GITHUB_SHA"
 
           if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
             # Triggered by build workflow completion
@@ -137,6 +139,7 @@ jobs:
             # Extract version info from commit message or use commit SHA
             # Use Git to generate consistent short SHA (ensures uniqueness like build.yml)
             short_sha=$(git rev-parse --short "$HEAD_SHA")
+            source_ref="$HEAD_SHA"
 
             # Determine build type based on triggering workflow event and ref
             triggering_event="$TRIGGERING_EVENT"
@@ -261,6 +264,23 @@ jobs:
                 echo "⚠️  Only release versions (latest, v1.0.0, 1.0.0) and prereleases (v1.0.0-alpha1, 1.0.0-beta2) are supported"
                 ;;
             esac
+
+            if [[ "$should_build" == true && "$input_version" != "latest" ]]; then
+              tag_ref="refs/tags/$input_version"
+              if ! git ls-remote --exit-code origin "$tag_ref" >/dev/null 2>&1; then
+                if [[ "$input_version" == v* ]]; then
+                  tag_ref="refs/tags/${input_version#v}"
+                else
+                  tag_ref="refs/tags/v$input_version"
+                fi
+              fi
+
+              if ! git ls-remote --exit-code origin "$tag_ref" >/dev/null 2>&1; then
+                echo "❌ Release tag not found for Docker build: $input_version"
+                exit 1
+              fi
+              source_ref="$tag_ref"
+            fi
           fi
 
           {
@@ -271,6 +291,7 @@ jobs:
             echo "short_sha=$short_sha"
             echo "is_prerelease=$is_prerelease"
             echo "create_latest=$create_latest"
+            echo "source_ref=$source_ref"
           } >> "$GITHUB_OUTPUT"
 
           echo "🐳 Docker Build Summary:"
@@ -281,6 +302,7 @@ jobs:
           echo "  - Short SHA: $short_sha"
           echo "  - Is prerelease: $is_prerelease"
           echo "  - Create latest: $create_latest"
+          echo "  - Source ref: $source_ref"
 
   # Build multi-arch Docker images
   # Strategy: Build images using pre-built binaries from dl.rustfs.com
@@ -308,6 +330,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: ${{ needs.build-check.outputs.source_ref }}
 
       - name: Login to Docker Hub
         uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3
@@ -397,7 +420,8 @@ jobs:
           LABELS="org.opencontainers.image.title=RustFS"
           LABELS="$LABELS,org.opencontainers.image.description=RustFS distributed object storage system"
           LABELS="$LABELS,org.opencontainers.image.version=$VERSION"
-          LABELS="$LABELS,org.opencontainers.image.revision=${{ github.sha }}"
+          SOURCE_REVISION="$(git rev-parse HEAD)"
+          LABELS="$LABELS,org.opencontainers.image.revision=$SOURCE_REVISION"
           LABELS="$LABELS,org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }}"
           LABELS="$LABELS,org.opencontainers.image.created=$(date -u +'%Y-%m-%dT%H:%M:%SZ')"
           LABELS="$LABELS,org.opencontainers.image.build-type=$BUILD_TYPE"

crates/kms/tests/behavior_backup.rs

@@ -225,8 +225,6 @@ async fn nothing_readable_leaves_the_bundle_unwrapped() {
             "artifact {} carries the raw on-disk record",
             artifact.path
         );
-        // A cheap structural check too: an encrypted payload is not JSON.
-        assert_ne!(payload.first(), Some(&b'{'), "artifact {} looks like plaintext JSON", artifact.path);
     }
 
     // The manifest itself is not encrypted, so assert directly that it carries

crates/utils/src/net.rs

@@ -659,9 +659,6 @@ mod test {
         // Port should be in valid range (u16 max is always <= 65535)
         assert!(port1 > 0);
         assert!(port2 > 0);
-
-        // Different calls should typically return different ports
-        assert_ne!(port1, port2);
     }
 
     #[test]

rustfs/src/admin/handlers/site_replication.rs

@@ -1067,9 +1067,13 @@ fn parse_site_replication_state(data: &[u8]) -> S3Result<SiteReplicationState> {
     state.peers = normalize_peer_map_by_identity(state.peers);
     // A peer-edit high-water mark only fences a CURRENT peer. A site that
     // leaves drops below two peers, which clears its own state object and
-    // restarts its generation counter at zero — a mark left over from the
-    // previous membership would then reject every edit it sends after it
-    // rejoins. Dropping departed origins on load also keeps the map bounded.
+    // restarts its generation counter — a mark left over from the previous
+    // membership must not reject the edits it sends after it rejoins. This
+    // pruning covers departures THIS site observed; an origin removed
+    // unilaterally elsewhere stays in this peer map with its mark, and the
+    // wall-clock floor in `next_peer_edit_generation` is what lifts its
+    // restarted counter over that mark. Dropping departed origins on load
+    // also keeps the map bounded.
     state
         .applied_edit_generations
         .retain(|origin, _| state.peers.contains_key(origin));
@@ -3000,6 +3004,9 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Fut
                 "admin site replication state"
             );
         }
+        // Failed peer deliveries recorded in the retry queue; runs behind the
+        // same lifecycle guard and pending_* gates as the reconcilers above.
+        drain_site_replication_retry_queue().await;
     })
 }
 
@@ -5935,11 +5942,51 @@ fn summarize_peer_error_detail(detail: &str) -> String {
     summary
 }
 
-/// Allocate the next peer-edit generation. Called inside the state
-/// transaction, so the counter is handed out under the distributed
-/// state-object lock and two nodes of this site can never take the same one.
+/// The wall clock in unix nanoseconds, clamped into u64. A pre-1970 (or
+/// post-2554) clock yields 0, which makes the hybrid allocation below
+/// degrade to the plain `previous + 1` counter — monotone, never panicking.
+fn edit_generation_wall_clock() -> u64 {
+    u64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos()).unwrap_or(0)
+}
+
+/// Allocate the next peer-edit generation as a hybrid logical clock:
+/// `max(wall clock in unix nanoseconds, previous + 1)`. Called inside the
+/// state transaction, so the value is handed out under the distributed
+/// state-object lock and two nodes of this site can never take the same one
+/// (`previous + 1` keeps the sequence strictly increasing even when two
+/// allocations land in one clock tick, and keeps it monotone on a node
+/// whose clock stepped backwards mid-lifetime).
+///
+/// The wall-clock floor is what survives the counter's death. A site
+/// removed while unreachable — the receiver never dropped it from its peer
+/// map, so the load-time mark pruning in `parse_site_replication_state`
+/// never fired — that later rejoins recreates its state object with the
+/// counter back at zero. A plain counter would then hand out generations
+/// below the receiver's stale high-water mark and every delivery would be
+/// silently fenced until the counter caught up. Jumping to wall time clears
+/// that mark: every value the deleted lifetime handed out was capped by the
+/// wall clock at its own allocation (or by a prior lifetime's cap, applied
+/// inductively), so the recreated lifetime's first allocation exceeds them
+/// all — while a pre-removal delivery still in flight stays below the new
+/// floor and remains correctly fenced. Marks recorded by pre-hybrid
+/// receivers (small plain-counter values) sit far below any wall-clock
+/// value, so a restarted origin passes those too — the fix needs only the
+/// sender upgraded, nothing on the wire or in the receiver changed.
+///
+/// A wall clock that regresses across a delete/recreate (the recreating
+/// node's clock behind the clock that fed the previous lifetime) mints
+/// below the stale mark and the origin stays fenced — but only until real
+/// time passes the previous lifetime's last allocation, because every later
+/// allocation takes the wall-clock floor again. Bounded by the skew,
+/// self-healing, and no rollback window beyond the plain counter's: a
+/// delivery applies only at or above the receiver's mark, so the one
+/// cross-lifetime interleaving that can apply stale content — a
+/// pre-removal delivery whose generation lands above everything the
+/// regressed new lifetime has minted — required the same straggler landing
+/// above the mark under the plain counter, where the recreated counter's
+/// low restart made it strictly easier to hit.
 fn next_peer_edit_generation(state: &mut SiteReplicationState) -> u64 {
-    state.edit_generation = state.edit_generation.saturating_add(1);
+    state.edit_generation = edit_generation_wall_clock().max(state.edit_generation.saturating_add(1));
     state.edit_generation
 }
 
@@ -6093,7 +6140,12 @@ async fn enqueue_site_replication_retry_event_for_generation(
     let path_owned = path.to_string();
     let error_text = error.to_string();
     let result = update_site_replication_state(move |state| {
-        upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
+        // A peer that left the state can never drain its entries again
+        // (remove_sites already pruned them); recording a late failure for it
+        // would only pollute retry_stats until the queue cap evicts it.
+        if state.peers.contains_key(&peer_owned.deployment_id) {
+            upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
+        }
         Ok(())
     })
     .await;
@@ -6127,6 +6179,396 @@ fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEvent) -> bool
     )
 }
 
+/// Exponential backoff base for the background retry drain, aligned with the
+/// reconcile cadence (`site_replication_reconcile::RECONCILE_INTERVAL`).
+const SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS: i64 = 600;
+/// Backoff ceiling: a permanently failed peer is still probed daily.
+const SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS: i64 = 86_400;
+
+/// What the background drain may do for one retry event. Everything not
+/// representable here is operator territory (manual repair).
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum RetryDrainAction {
+    /// Constant-path IAM item deliveries collapse into one queue entry per
+    /// peer and their bodies are not persisted; the only faithful replay is
+    /// the current IAM snapshot from the bootstrap plan.
+    IamSnapshot,
+    /// Same collapse for bucket-meta deliveries: replay the bucket metadata
+    /// snapshot from the bootstrap plan.
+    BucketMetadataSnapshot,
+    /// A self-contained bucket op the bootstrap plan can re-derive for its
+    /// bucket (`make-with-versioning` / `configure-replication`).
+    BucketOpReplay { operation: String, bucket: String },
+    /// Re-send the current peer records under a fresh edit generation.
+    PeerEdit,
+}
+
+fn classify_site_replication_retry_event(event: &SiteReplicationRetryEvent) -> Option<RetryDrainAction> {
+    if event.path.starts_with("internal:") {
+        // Marker records store payloads in `last_error` (legacy
+        // pending-endpoint-refresh backup); they are not delivery failures.
+        return None;
+    }
+    let base_path = event.path.split_once('?').map(|(base, _)| base).unwrap_or(&event.path);
+    match base_path {
+        "/rustfs/admin/v3/site-replication/peer/iam-item" => Some(RetryDrainAction::IamSnapshot),
+        "/rustfs/admin/v3/site-replication/peer/bucket-meta" => Some(RetryDrainAction::BucketMetadataSnapshot),
+        SITE_REPLICATION_PEER_EDIT_PATH => Some(RetryDrainAction::PeerEdit),
+        SITE_REPLICATION_PEER_BUCKET_OPS_PATH => {
+            let operation = retry_bucket_operation(&event.path)?;
+            if !matches!(
+                operation.as_str(),
+                SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING | SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION
+            ) {
+                // Destructive ops (delete-bucket / force-delete-bucket) are
+                // operator territory: replaying them against a peer whose
+                // bucket was since recreated is irreversible.
+                return None;
+            }
+            let bucket = retry_bucket_name(&event.path)?;
+            Some(RetryDrainAction::BucketOpReplay { operation, bucket })
+        }
+        _ => None,
+    }
+}
+
+fn retry_bucket_name(path: &str) -> Option<String> {
+    let (_, query) = path.split_once('?')?;
+    form_urlencoded::parse(query.as_bytes())
+        .find_map(|(key, value)| (key == "bucket" && !value.is_empty()).then(|| value.into_owned()))
+}
+
+/// Settle a constant-path (collapsed) retry event only if no newer failure
+/// was recorded for it after `snapshot_updated_at`. The drain's snapshot
+/// resend proves delivery of the state as of plan-build time; a failure
+/// recorded during the (potentially long) delivery window belongs to a newer
+/// local commit the snapshot did not contain, and clearing it would leave the
+/// peer silently diverged. Returns how many events would be (or were) kept
+/// back for that reason plus how many were settled.
+fn settle_site_replication_retry_events_up_to(
+    queue: &mut Vec<SiteReplicationRetryEvent>,
+    peer: &PeerInfo,
+    path: &str,
+    snapshot_updated_at: Option<OffsetDateTime>,
+) -> usize {
+    let before = queue.len();
+    queue.retain(|event| {
+        if !retry_event_matches(event, peer, path) {
+            return true;
+        }
+        match (event.updated_at, snapshot_updated_at) {
+            // A failure stamped after our snapshot: keep it.
+            (Some(current), Some(seen)) => current > seen,
+            (Some(_), None) => true,
+            // Legacy entry without a timestamp cannot be newer than anything.
+            (None, _) => false,
+        }
+    });
+    before - queue.len()
+}
+
+async fn dequeue_site_replication_retry_event_up_to(peer: &PeerInfo, path: &str, snapshot_updated_at: Option<OffsetDateTime>) {
+    let peer_owned = peer.clone();
+    let path_owned = path.to_string();
+    let result = update_site_replication_state(move |state| {
+        settle_site_replication_retry_events_up_to(&mut state.retry_queue, &peer_owned, &path_owned, snapshot_updated_at);
+        Ok(())
+    })
+    .await;
+
+    if let Err(err) = result {
+        warn!(
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            peer = %peer.endpoint,
+            deployment_id = %peer.deployment_id,
+            path,
+            error = ?err,
+            "failed to settle site replication retry event"
+        );
+    }
+}
+
+/// Whether the drain may attempt this event now.
+fn site_replication_retry_backoff_elapsed(event: &SiteReplicationRetryEvent, now: OffsetDateTime) -> bool {
+    let Some(updated_at) = event.updated_at else {
+        return true;
+    };
+    // 600 * 2^8 already exceeds the daily ceiling; capping the shift keeps
+    // the arithmetic overflow-free for any persisted retry_count.
+    let exponent = event.retry_count.saturating_sub(1).min(8);
+    let delay = (SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS << exponent).min(SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS);
+    now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= delay
+}
+
+/// The subset of the retry queue the background drain is allowed to touch.
+fn actionable_site_replication_retry_events(state: &SiteReplicationState, now: OffsetDateTime) -> Vec<SiteReplicationRetryEvent> {
+    state
+        .retry_queue
+        .iter()
+        .filter(|event| classify_site_replication_retry_event(event).is_some())
+        .filter(|event| state.peers.contains_key(&event.peer_deployment_id))
+        .filter(|event| site_replication_retry_backoff_elapsed(event, now))
+        .cloned()
+        .collect()
+}
+
+/// Background consumer for the retry queue, run from the reconcile tick.
+///
+/// Scope: this settles "delivered once and failed" entries. A hook that never
+/// fired (crash between the local commit and the send) leaves no entry, so
+/// the drain is not a full cross-site diff-heal; manual repair remains the
+/// authoritative catch-all.
+async fn drain_site_replication_retry_queue() {
+    if let Err(err) = drain_site_replication_retry_queue_inner().await {
+        warn!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "retry_drain_failed",
+            error = ?err,
+            "admin site replication state"
+        );
+    }
+}
+
+async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
+    let Some(runtime) = runtime_site_replication_targets().await? else {
+        return Ok(());
+    };
+    let actionable = actionable_site_replication_retry_events(&runtime.state, OffsetDateTime::now_utc());
+    if actionable.is_empty() {
+        return Ok(());
+    }
+    let Some(store) = current_object_store_handle() else {
+        return Ok(());
+    };
+    if runtime.state.pending_endpoint_refresh.is_some()
+        || runtime.state.pending_remove.is_some()
+        || runtime.state.pending_rotation.is_some()
+    {
+        // The tick-level gate ran before the reconcilers; a multi-step flow
+        // (endpoint refresh commits its pending marker without the lifecycle
+        // guard) may have started since. Re-check on the fresh state.
+        return Ok(());
+    }
+    // Serialize against operator repair execution. This does NOT close the
+    // dry-run -> execute window (dry-run takes no lock): a drain settling a
+    // replayable bucket-op entry in that window changes the preflight token
+    // and execute fails safe with "preflight is stale" — the operator
+    // re-runs the dry-run. Lock order matches repair: lifecycle guard (held
+    // by the reconcile tick) -> repair execution lock -> state object lock
+    // inside the send bookkeeping. An operator repair holding the lock makes
+    // this tick skip after the lock-acquire timeout.
+    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
+        drain_site_replication_retry_queue_locked(runtime, actionable).await
+    })
+    .await
+    .map_err(ApiError::from)?
+}
+
+async fn drain_site_replication_retry_queue_locked(
+    runtime: SiteReplicationRuntime,
+    events: Vec<SiteReplicationRetryEvent>,
+) -> S3Result<()> {
+    let needs_plan = events
+        .iter()
+        .any(|event| !matches!(classify_site_replication_retry_event(event), Some(RetryDrainAction::PeerEdit)));
+    // The plan is a full local snapshot (buckets + IAM); build it once per
+    // tick and only when a snapshot resend is actually due.
+    let plan = if needs_plan {
+        let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
+        Some(site_replication_bootstrap_plan(&info)?)
+    } else {
+        None
+    };
+
+    let mut events_by_peer: BTreeMap<String, Vec<SiteReplicationRetryEvent>> = BTreeMap::new();
+    for event in events {
+        events_by_peer
+            .entry(event.peer_deployment_id.clone())
+            .or_default()
+            .push(event);
+    }
+
+    let mut settled = 0usize;
+    let mut failures = 0usize;
+    for (deployment_id, peer_events) in events_by_peer {
+        let Some(peer) = runtime.state.peers.get(&deployment_id) else {
+            continue;
+        };
+        if deployment_id == runtime.local_peer.deployment_id
+            || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        {
+            continue;
+        }
+        let transport = match PeerTransport::for_runtime_peer(peer).await {
+            Ok(transport) => transport,
+            Err(err) => {
+                // Record the attempt so backoff advances for an unreachable
+                // peer instead of re-dialing it every tick.
+                for event in &peer_events {
+                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                }
+                failures += peer_events.len();
+                continue;
+            }
+        };
+        for event in peer_events {
+            let Some(action) = classify_site_replication_retry_event(&event) else {
+                continue;
+            };
+            match drain_one_site_replication_retry_event(&runtime, peer, &transport, &event, action, plan.as_ref()).await {
+                Ok(true) => settled += 1,
+                Ok(false) => {}
+                Err(_) => failures += 1,
+            }
+        }
+    }
+
+    if settled > 0 || failures > 0 {
+        info!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "retry_drain_settled",
+            settled,
+            failures,
+            "admin site replication state"
+        );
+    }
+    Ok(())
+}
+
+/// Replay one retry event against its peer. Returns `Ok(true)` when the
+/// event was settled (delivered, or provably stale), `Ok(false)` when it was
+/// skipped, and `Err` after a failed delivery (already re-queued with an
+/// incremented retry count).
+async fn drain_one_site_replication_retry_event(
+    runtime: &SiteReplicationRuntime,
+    peer: &PeerInfo,
+    transport: &PeerTransport,
+    event: &SiteReplicationRetryEvent,
+    action: RetryDrainAction,
+    plan: Option<&SiteReplicationBootstrapPlan>,
+) -> S3Result<bool> {
+    let access_key = &runtime.state.service_account_access_key;
+    let secret_key = &runtime.service_account_secret_key;
+    match action {
+        RetryDrainAction::IamSnapshot | RetryDrainAction::BucketMetadataSnapshot => {
+            let Some(plan) = plan else {
+                return Ok(false);
+            };
+            let tasks: Vec<SiteReplicationRepairTask<'_>> = match action {
+                RetryDrainAction::IamSnapshot => plan.iam_items.iter().map(SiteReplicationRepairTask::Iam).collect(),
+                _ => plan
+                    .bucket_items
+                    .iter()
+                    .map(SiteReplicationRepairTask::BucketMetadata)
+                    .collect(),
+            };
+            for task in &tasks {
+                if let Err(err) = task.send(transport, access_key, secret_key).await {
+                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                    return Err(err);
+                }
+            }
+            // Conditional settlement: iam-item / bucket-meta entries collapse
+            // per (peer, path), so a hook failure recorded while this snapshot
+            // was in flight belongs to a commit the snapshot did not contain
+            // and must survive this success.
+            dequeue_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await;
+            Ok(true)
+        }
+        RetryDrainAction::BucketOpReplay { operation, bucket } => {
+            let Some(plan) = plan else {
+                return Ok(false);
+            };
+            // Replay from the CURRENT plan, never the recorded path: the
+            // recorded query can carry an expired one-shot bootstrap token or
+            // a stale createdAt.
+            let make_op = operation == SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING;
+            let paths = if make_op {
+                &plan.bucket_make_ops
+            } else {
+                &plan.bucket_configure_ops
+            };
+            let tasks: Vec<SiteReplicationRepairTask<'_>> = paths
+                .iter()
+                .filter(|path| retry_bucket_name(path).as_deref() == Some(bucket.as_str()))
+                .map(|path| {
+                    if make_op {
+                        SiteReplicationRepairTask::BucketMake(path)
+                    } else {
+                        SiteReplicationRepairTask::Replication(path)
+                    }
+                })
+                .collect();
+            if tasks.is_empty() {
+                // The bucket left the plan (deleted, or replication no longer
+                // configured): the recorded intent is stale, settle it.
+                dequeue_site_replication_retry_event(peer, &event.path).await;
+                return Ok(true);
+            }
+            for task in &tasks {
+                if let Err(err) = task.send(transport, access_key, secret_key).await {
+                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                    return Err(err);
+                }
+            }
+            dequeue_site_replication_retry_event(peer, &event.path).await;
+            Ok(true)
+        }
+        RetryDrainAction::PeerEdit => {
+            // The recorded generation is stale by definition — the receiver
+            // fences it. Allocate a fresh generation and re-send the current
+            // peer records (a superset of the failed body; the receiver
+            // upserts), all inside one state transaction so the fence and the
+            // bodies agree.
+            let target_id = peer.deployment_id.clone();
+            let (generation, bodies) = update_site_replication_state(move |state| {
+                if !state.peers.contains_key(&target_id) {
+                    return Ok((None, Vec::new()));
+                }
+                Ok((Some(next_peer_edit_generation(state)), state.peers.values().cloned().collect::<Vec<_>>()))
+            })
+            .await?;
+            let Some(generation) = generation else {
+                // Peer left between the snapshot and now; the queue entry was
+                // already pruned by remove_sites.
+                return Ok(false);
+            };
+            let local_deployment_id = Some(runtime.local_peer.deployment_id.as_str()).filter(|id| !id.is_empty());
+            let edit_path = peer_edit_path_with_fence(local_deployment_id, generation);
+            let delivery_fence = local_deployment_id.is_some().then_some(generation);
+            for body in &bodies {
+                if let Err(err) = send_peer_admin_request_with_client(
+                    &transport.client,
+                    &transport.connection,
+                    &edit_path,
+                    access_key,
+                    secret_key,
+                    body,
+                )
+                .await
+                {
+                    enqueue_site_replication_retry_event_for_generation(
+                        peer,
+                        SITE_REPLICATION_PEER_EDIT_PATH,
+                        &err,
+                        delivery_fence,
+                    )
+                    .await;
+                    return Err(err);
+                }
+            }
+            dequeue_site_replication_retry_event_for_generation(peer, SITE_REPLICATION_PEER_EDIT_PATH, delivery_fence).await;
+            Ok(true)
+        }
+    }
+}
+
 /// Remove a retry event for (peer, path) from the queue on successful delivery.
 /// This is a no-op (load + no-op persist skipped) when no matching entry exists,
 /// avoiding unnecessary I/O on the common path.
@@ -11383,6 +11825,189 @@ mod tests {
         assert!(target_state.peers["remote"].skip_tls_verify);
     }
 
+    fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option<OffsetDateTime>) -> SiteReplicationRetryEvent {
+        SiteReplicationRetryEvent {
+            id: format!("evt-{peer}"),
+            peer_deployment_id: peer.to_string(),
+            peer_endpoint: format!("https://{peer}.example.com"),
+            path: path.to_string(),
+            retry_count,
+            failed: retry_count >= SITE_REPLICATION_RETRY_FAILED_AFTER,
+            last_error: "remote-operation-failed".to_string(),
+            updated_at,
+            edit_generation: None,
+        }
+    }
+
+    /// P1-3 red-light: the drain must only ever act on deliveries it can
+    /// replay faithfully. IAM / bucket-meta entries collapse per (peer, path)
+    /// with no body persisted — only a snapshot resend is truthful; bucket
+    /// makes/replication configs are re-derivable; destructive bucket ops and
+    /// `internal:` marker records (the pending-endpoint-refresh backup store)
+    /// are never background-replayed.
+    #[test]
+    fn test_classify_site_replication_retry_event_actions() {
+        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let classify = |path: &str| classify_site_replication_retry_event(&drain_event("remote", path, 1, Some(now)));
+
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/iam-item"),
+            Some(RetryDrainAction::IamSnapshot)
+        );
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-meta"),
+            Some(RetryDrainAction::BucketMetadataSnapshot)
+        );
+        assert_eq!(classify(SITE_REPLICATION_PEER_EDIT_PATH), Some(RetryDrainAction::PeerEdit));
+        assert_eq!(
+            classify(
+                "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning&createdAt=1"
+            ),
+            Some(RetryDrainAction::BucketOpReplay {
+                operation: SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING.to_string(),
+                bucket: "photos".to_string(),
+            })
+        );
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=configure-replication"),
+            Some(RetryDrainAction::BucketOpReplay {
+                operation: SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION.to_string(),
+                bucket: "photos".to_string(),
+            })
+        );
+        // Destructive ops are operator territory: replaying a bucket delete
+        // against a peer whose bucket was since recreated is irreversible.
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket"),
+            None
+        );
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=force-delete-bucket"),
+            None
+        );
+        // `internal:` records store payloads in `last_error`, not failures.
+        assert_eq!(classify(SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH), None);
+        assert_eq!(classify("internal:some-future-marker"), None);
+        assert_eq!(classify("/rustfs/admin/v3/site-replication/peer/unknown"), None);
+    }
+
+    /// Exponential backoff gates every attempt: without it a dead peer's
+    /// entries hit `failed` (retry_count >= 3) within 30 minutes of reconcile
+    /// ticks and the retry stats lose their signal.
+    #[test]
+    fn test_site_replication_retry_backoff_schedule() {
+        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let at = |secs_ago: i64| Some(now - time::Duration::seconds(secs_ago));
+        let elapsed = |retry_count: u32, secs_ago: i64| {
+            site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", retry_count, at(secs_ago)), now)
+        };
+
+        // No record of when it failed: attempt now.
+        assert!(site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", 1, None), now));
+        // First failure: one reconcile interval.
+        assert!(!elapsed(1, 599));
+        assert!(elapsed(1, 601));
+        // Third failure: 600 * 2^2 = 2400s.
+        assert!(!elapsed(3, 1200));
+        assert!(elapsed(3, 2401));
+        // Ceiling: a long-dead peer is still probed daily, never less often.
+        assert!(!elapsed(30, 86_000));
+        assert!(elapsed(30, 86_401));
+    }
+
+    /// The actionable subset respects classification, peer membership and
+    /// backoff; everything else stays untouched in the queue.
+    #[test]
+    fn test_actionable_site_replication_retry_events_filters() {
+        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let old = Some(now - time::Duration::seconds(700));
+        let mut state = SiteReplicationState::default();
+        state
+            .peers
+            .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
+
+        state.retry_queue = vec![
+            // Eligible: known peer, replayable, past backoff.
+            drain_event("remote", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old),
+            // Not yet due.
+            drain_event("remote", "/rustfs/admin/v3/site-replication/peer/bucket-meta", 2, Some(now)),
+            // Unknown peer (removed since the failure was recorded).
+            drain_event("gone", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old),
+            // Marker record, not a delivery failure.
+            drain_event("remote", SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH, 0, old),
+            // Destructive op: operator-only.
+            drain_event(
+                "remote",
+                "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket",
+                1,
+                old,
+            ),
+        ];
+
+        let actionable = actionable_site_replication_retry_events(&state, now);
+        assert_eq!(actionable.len(), 1, "only the due, replayable, known-peer event is actionable");
+        assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/iam-item");
+    }
+
+    /// The drain settles a peer-edit success under a freshly allocated
+    /// generation; legacy queue entries carry `edit_generation: None` and
+    /// must be cleared by that generation-scoped settlement (`(Some, None)`
+    /// falls through to removal), or the drain would spin on them forever.
+    #[test]
+    fn test_settle_clears_legacy_none_generation_event_for_generation_scoped_success() {
+        let target = peer("remote", "https://remote.example.com");
+        let mut queue = vec![drain_event("remote", SITE_REPLICATION_PEER_EDIT_PATH, 1, None)];
+        assert!(queue[0].edit_generation.is_none());
+
+        let settled = settle_site_replication_retry_events(&mut queue, &target, SITE_REPLICATION_PEER_EDIT_PATH, Some(42));
+
+        assert_eq!(settled, 1, "a legacy None-generation event must settle under a newer generation");
+        assert!(queue.is_empty());
+    }
+
+    /// Conditional settlement for collapsed (constant-path) entries: a
+    /// failure stamped after the drain snapshot belongs to a newer local
+    /// commit the snapshot did not contain and must survive the snapshot's
+    /// success.
+    #[test]
+    fn test_settle_up_to_keeps_failures_newer_than_the_snapshot() {
+        let target = peer("remote", "https://remote.example.com");
+        let path = "/rustfs/admin/v3/site-replication/peer/iam-item";
+        let snapshot_at = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+
+        // Failure re-stamped after the snapshot: kept.
+        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at + time::Duration::seconds(5)))];
+        assert_eq!(
+            settle_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            0
+        );
+        assert_eq!(queue.len(), 1, "a failure newer than the snapshot must survive");
+
+        // Unchanged since the snapshot: settled.
+        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at))];
+        assert_eq!(
+            settle_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            1
+        );
+        assert!(queue.is_empty());
+
+        // Legacy entry without a timestamp cannot be newer: settled.
+        let mut queue = vec![drain_event("remote", path, 2, None)];
+        assert_eq!(
+            settle_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            1
+        );
+        assert!(queue.is_empty());
+
+        // Other (peer, path) entries are untouched.
+        let mut queue = vec![drain_event("other", path, 2, Some(snapshot_at))];
+        assert_eq!(
+            settle_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            0
+        );
+        assert_eq!(queue.len(), 1);
+    }
+
     #[test]
     fn test_pending_endpoint_refresh_retry_summary_redacts_pem() {
         let pem = "-----BEGIN CERTIFICATE-----\nsecret-marker\n-----END CERTIFICATE-----";
@@ -13244,6 +13869,104 @@ mod tests {
         assert!(!peer_edit_delivery_is_stale(&reloaded, "origin-site", 1));
     }
 
+    /// The unilateral-removal rejoin gap the hybrid clock closes. The origin
+    /// was removed while unreachable, but THIS site never dropped it from
+    /// its peer map, so the load-time mark pruning never fired and the mark
+    /// from the previous membership survives. The origin's recreated state
+    /// object restarts its counter, and with a plain `previous + 1` counter
+    /// every delivery it sent — generations 1, 2, … below the stale mark —
+    /// would be silently acked-and-dropped until the counter caught up. The
+    /// wall-clock floor in `next_peer_edit_generation` lifts the restarted
+    /// counter over every value the deleted lifetime handed out. Reverting
+    /// the allocation to the plain counter (dropping the wall-clock max)
+    /// turns the not-stale assertion red.
+    #[test]
+    fn hybrid_generation_unfences_a_rejoined_origin_whose_counter_restarted() {
+        // First lifetime of the origin's state object: two allocations, both
+        // capped by the wall clock at their own allocation.
+        let mut first_life = SiteReplicationState::default();
+        let straggler = next_peer_edit_generation(&mut first_life);
+        let last_applied = next_peer_edit_generation(&mut first_life);
+        assert!(last_applied > straggler, "allocations must be strictly increasing");
+
+        // The receiver applied up to `last_applied` and keeps the origin in
+        // its peer map across the unilateral removal — reloading must keep
+        // the mark, which is exactly why pruning cannot cover this case.
+        let mut receiver = SiteReplicationState::default();
+        receiver.peers.insert(
+            "origin-site".to_string(),
+            PeerInfo {
+                deployment_id: "origin-site".to_string(),
+                ..peer("origin", "https://origin.example:9000")
+            },
+        );
+        record_applied_peer_edit_generation(&mut receiver, "origin-site", last_applied);
+        let mut receiver = parse_site_replication_state(&serde_json::to_vec(&receiver).expect("serialize")).expect("reload");
+        assert_eq!(receiver.applied_edit_generations.get("origin-site"), Some(&last_applied));
+
+        // The origin rejoins with a RECREATED state object: counter back at
+        // zero. The wall-clock floor must lift its first allocation over the
+        // previous lifetime's mark…
+        let mut second_life = SiteReplicationState::default();
+        let restarted = next_peer_edit_generation(&mut second_life);
+        assert!(
+            !peer_edit_delivery_is_stale(&receiver, "origin-site", restarted),
+            "the recreated lifetime's first allocation ({restarted}) must not be fenced by the previous lifetime's mark ({last_applied})"
+        );
+        record_applied_peer_edit_generation(&mut receiver, "origin-site", restarted);
+
+        // …while a pre-removal delivery still in flight stays below the new
+        // floor and remains correctly fenced — the rollback the fence exists
+        // to reject.
+        assert!(
+            peer_edit_delivery_is_stale(&receiver, "origin-site", straggler),
+            "a pre-removal in-flight delivery ({straggler}) must stay fenced after the rejoin"
+        );
+    }
+
+    /// Marks recorded before the hybrid clock existed are small plain-counter
+    /// values, far below any wall-clock allocation: a restarted origin passes
+    /// them as soon as the SENDER runs the hybrid clock — nothing changes on
+    /// the wire or in the receiver, so pre-hybrid receivers get the fix too.
+    /// The other direction is unchanged: among plain-counter values the
+    /// generation order still fences the delivery that lost the race.
+    #[test]
+    fn hybrid_generation_passes_marks_recorded_by_plain_counter_receivers() {
+        let mut receiver = SiteReplicationState::default();
+        record_applied_peer_edit_generation(&mut receiver, "origin-site", 57);
+        assert!(peer_edit_delivery_is_stale(&receiver, "origin-site", 56));
+        assert!(!peer_edit_delivery_is_stale(&receiver, "origin-site", 57));
+
+        let mut rejoined = SiteReplicationState::default();
+        let restarted = next_peer_edit_generation(&mut rejoined);
+        assert!(
+            !peer_edit_delivery_is_stale(&receiver, "origin-site", restarted),
+            "a wall-clock allocation ({restarted}) must clear a plain-counter mark (57)"
+        );
+    }
+
+    /// The `previous + 1` half of the hybrid clock: allocations stay strictly
+    /// increasing even when the wall clock cannot move them forward — two
+    /// allocations inside one clock tick, or a clock that stepped backwards
+    /// mid-lifetime (a counter already ahead of the wall clock advances by
+    /// exactly one per allocation instead of jumping back). Dropping the
+    /// `previous + 1` half (allocating bare wall time) turns this red.
+    #[test]
+    fn hybrid_generation_is_strictly_increasing_when_the_clock_stalls() {
+        let mut state = SiteReplicationState {
+            // A counter far ahead of any wall clock this test will see.
+            edit_generation: u64::MAX / 2,
+            ..Default::default()
+        };
+        assert_eq!(next_peer_edit_generation(&mut state), u64::MAX / 2 + 1);
+        assert_eq!(next_peer_edit_generation(&mut state), u64::MAX / 2 + 2);
+        // Saturation pins at the ceiling instead of wrapping; the equal-value
+        // escape (`applied > generation` is false for equal) keeps deliveries
+        // applying rather than fencing the origin out.
+        state.edit_generation = u64::MAX;
+        assert_eq!(next_peer_edit_generation(&mut state), u64::MAX);
+    }
+
     #[test]
     fn test_retry_stats_for_state_counts_pending_and_failed() {
         let state = SiteReplicationState {
@@ -16044,10 +16767,77 @@ mod tests {
             generations.len(),
             "two nodes took the same edit generation, so their deliveries cannot be ordered: {generations:?}"
         );
+        // The hybrid clock allocates `max(wall nanos, previous + 1)` — the
+        // persisted counter is the largest allocation, and the `+ 1` half
+        // keeps allocations distinct even inside one clock tick.
+        assert_eq!(
+            Some(&load_site_replication_state().await.expect("reload").edit_generation),
+            unique.last(),
+            "the persisted counter must be the largest allocation handed out"
+        );
+    }
+
+    /// The unilateral-removal rejoin, end to end across the state object's
+    /// real lifecycle: dropping below two peers clears the object (the
+    /// counter dies with it), and the recreated object's first allocation —
+    /// raced by two nodes — must clear the previous lifetime's values via
+    /// the wall-clock floor, so a receiver still holding the old mark
+    /// accepts the restarted counter instead of fencing it.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_recreated_state_object_allocates_over_the_previous_lifetimes_mark() {
+        publish_ready_iam_context().await;
+        let seed = || SiteReplicationState {
+            peers: ["site-a", "site-b"]
+                .into_iter()
+                .map(|name| (name.to_string(), peer(name, &format!("https://{name}.example:9000"))))
+                .collect(),
+            ..Default::default()
+        };
+
+        save_site_replication_state(&seed()).await.expect("seed state");
+        let straggler = update_site_replication_state(|state| Ok(next_peer_edit_generation(state)))
+            .await
+            .expect("first-life allocation");
+        let last_applied = update_site_replication_state(|state| Ok(next_peer_edit_generation(state)))
+            .await
+            .expect("first-life allocation");
+        // A receiver that never dropped this site from its peer map holds
+        // this mark across the removal.
+        let mut receiver = SiteReplicationState::default();
+        record_applied_peer_edit_generation(&mut receiver, "origin-site", last_applied);
+
+        // Unilateral removal: the site drops below two peers, which clears
+        // its state object and the counter with it.
+        let mut departed = seed();
+        departed.peers.remove("site-b");
+        save_site_replication_state(&departed).await.expect("clear state");
         assert_eq!(
             load_site_replication_state().await.expect("reload").edit_generation,
-            generations.len() as u64,
-            "the persisted counter must account for every allocation"
+            0,
+            "clearing the state object must take the counter with it"
+        );
+
+        // Rejoin recreates the state object; two nodes race the first
+        // allocation of the new life.
+        save_site_replication_state(&seed()).await.expect("recreate state");
+        let node_a = tokio::spawn(update_site_replication_state(|state| Ok(next_peer_edit_generation(state))));
+        let node_b = tokio::spawn(update_site_replication_state(|state| Ok(next_peer_edit_generation(state))));
+        let generation_a = node_a.await.expect("node a task").expect("node a allocation");
+        let generation_b = node_b.await.expect("node b task").expect("node b allocation");
+        assert_ne!(generation_a, generation_b, "racing allocations must stay distinct");
+
+        // The receiver's stale mark must not fence the restarted counter…
+        let restarted = generation_a.min(generation_b);
+        assert!(
+            !peer_edit_delivery_is_stale(&receiver, "origin-site", restarted),
+            "the recreated life's first allocation ({restarted}) must clear the previous life's mark ({last_applied})"
+        );
+        record_applied_peer_edit_generation(&mut receiver, "origin-site", restarted);
+        // …while the cleared life's in-flight leftovers stay fenced.
+        assert!(
+            peer_edit_delivery_is_stale(&receiver, "origin-site", straggler),
+            "a pre-removal in-flight delivery ({straggler}) must stay fenced after the rejoin"
         );
     }
 
@@ -16063,17 +16853,31 @@ mod tests {
     async fn test_retry_event_persist_must_not_wipe_concurrent_locked_rmw() {
         publish_ready_iam_context().await;
 
+        const ROUNDS: usize = 8;
         let seed = SiteReplicationState {
             pending_rotation: Some(PendingRotation {
                 id: "rot-1".to_string(),
                 access_key: "svc-account".to_string(),
                 ..Default::default()
             }),
+            // Retry events are only recorded for current peers; seed them so
+            // the concurrency assertion below exercises the persist path.
+            peers: (0..ROUNDS)
+                .map(|round| {
+                    let deployment_id = format!("peer-{round}-deployment");
+                    (
+                        deployment_id.clone(),
+                        PeerInfo {
+                            endpoint: format!("https://peer-{round}.example:9000"),
+                            deployment_id,
+                            ..Default::default()
+                        },
+                    )
+                })
+                .collect(),
             ..Default::default()
         };
         save_site_replication_state(&seed).await.expect("seed state");
-
-        const ROUNDS: usize = 8;
         for round in 0..ROUNDS {
             let peer = PeerInfo {
                 endpoint: format!("https://peer-{round}.example:9000"),

scripts/security/check_preview_release_workflow.sh

@@ -195,6 +195,13 @@ IFS= read -r -d '' expected_docker_automatic_guard <<'EOF' || true
 EOF
 expected_docker_automatic_guard=${expected_docker_automatic_guard%$'\n'}
 require_job_if "$docker_workflow" "build-check" "$expected_docker_automatic_guard"
+require_line "$docker_workflow" '      source_ref: ${{ steps.check.outputs.source_ref }}' "Docker source ref output"
+require_line "$docker_workflow" '            source_ref="$HEAD_SHA"' "automatic Docker source ref"
+require_line "$docker_workflow" '              source_ref="$tag_ref"' "manual Docker source ref"
+require_line "$docker_workflow" '          ref: ${{ needs.build-check.outputs.source_ref }}' "Docker release source checkout"
+require_line "$docker_workflow" '          SOURCE_REVISION="$(git rev-parse HEAD)"' "Docker source revision resolution"
+require_line "$docker_workflow" '          LABELS="$LABELS,org.opencontainers.image.revision=$SOURCE_REVISION"' "Docker revision label"
+require_absent "$docker_workflow" 'org.opencontainers.image.revision=${{ github.sha }}' "Docker revision must not use the workflow branch SHA"
 
 docker_manual_guard=$(awk '
   $0 == "              *-preview*)" { in_preview = 1 }