feat(ecstore): report inline early-stop miss reasons (#6134)

Co-authored-by: heihutu <heihutu@gmail.com>

.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/ecstore/src/diagnostics/get.rs

@@ -190,6 +190,17 @@ pub(crate) const GET_METADATA_CACHE_REASON_VERSION_SUSPENDED: &str = "version_su
 pub(crate) const GET_METADATA_CACHE_REASON_VERSIONED: &str = "versioned";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA: &str = "conflicting_metadata";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER: &str = "delete_marker";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY: &str = "data_read_inline_body_verify";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED: &str = "data_read_inline_deleted";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY: &str = "data_read_inline_geometry";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH: &str = "data_read_inline_identity_mismatch";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD: &str = "data_read_inline_missing_payload";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD: &str = "data_read_inline_missing_shard";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE: &str = "data_read_inline_not_inline";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE: &str = "data_read_inline_part_shape";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE: &str = "data_read_inline_remote";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE: &str = "data_read_inline_size";
+pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED: &str = "data_read_inline_transformed";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_ERROR: &str = "error";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM: &str = "insufficient_quorum";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_NOT_FOUND: &str = "not_found";
@@ -551,6 +562,32 @@ mod tests {
         assert_eq!(GET_METADATA_CACHE_REASON_VERSIONED, "versioned");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, "conflicting_metadata");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, "delete_marker");
+        assert_eq!(
+            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
+            "data_read_inline_body_verify"
+        );
+        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, "data_read_inline_deleted");
+        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY, "data_read_inline_geometry");
+        assert_eq!(
+            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
+            "data_read_inline_identity_mismatch"
+        );
+        assert_eq!(
+            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
+            "data_read_inline_missing_payload"
+        );
+        assert_eq!(
+            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD,
+            "data_read_inline_missing_shard"
+        );
+        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE, "data_read_inline_not_inline");
+        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, "data_read_inline_part_shape");
+        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE, "data_read_inline_remote");
+        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, "data_read_inline_size");
+        assert_eq!(
+            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
+            "data_read_inline_transformed"
+        );
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_ERROR, "error");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, "insufficient_quorum");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, "not_found");

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -32,15 +32,22 @@ use crate::diagnostics::get::{
     GET_METADATA_CACHE_REASON_NOT_READ_DATA, GET_METADATA_CACHE_REASON_PART_NUMBER,
     GET_METADATA_CACHE_REASON_RAW_DATA_MOVEMENT_READ, GET_METADATA_CACHE_REASON_USABLE, GET_METADATA_CACHE_REASON_VERSION_ID,
     GET_METADATA_CACHE_REASON_VERSION_SUSPENDED, GET_METADATA_CACHE_REASON_VERSIONED,
-    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER,
+    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
-    GET_METADATA_EARLY_STOP_REASON_ERROR, GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
-    GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
-    GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
-    GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND, GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE,
-    GET_METADATA_RESPONSE_ERROR, GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE,
-    GET_METADATA_RESPONSE_VALID, GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
-    GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK,
+    GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, GET_METADATA_EARLY_STOP_REASON_ERROR,
-    GET_STAGE_DECODE, GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
+    GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, GET_METADATA_EARLY_STOP_REASON_NOT_FOUND,
+    GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND,
+    GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND, GET_METADATA_RESPONSE_ERROR,
+    GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT, GET_METADATA_RESPONSE_VALID,
+    GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_DIRECT_MEMORY,
+    GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE,
+    GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
     GET_STAGE_READER_SETUP_DROP_PENDING, GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM,
     GET_STAGE_READER_TASK_BITROT_READER_INIT, GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION,
     GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
@@ -652,36 +659,48 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
         && left.erasure.distribution == right.erasure.distribution
 }
 
-pub(in crate::set_disk) async fn data_read_early_stop_inline_body_verified(
+pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     bucket: &str,
     object: &str,
     candidate: &FileInfo,
     parts_metadata: &[FileInfo],
     disks: &[Option<DiskStore>],
-) -> bool {
+) -> Option<&'static str> {
-    if !candidate.inline_data()
+    if !candidate.inline_data() {
-        || candidate.is_compressed()
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
+    }
+    if candidate.is_compressed()
         || candidate
             .metadata
             .keys()
             .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
-        || candidate.is_remote()
-        || candidate.deleted
-        || candidate.size <= 0
-        || candidate.parts.len() != 1
-        || !candidate.has_valid_erasure_geometry()
     {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED);
+    }
+    if candidate.is_remote() {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE);
+    }
+    if candidate.deleted {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED);
+    }
+    if candidate.size <= 0 {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
+    }
+    if candidate.parts.len() != 1 {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+    }
+    if !candidate.has_valid_erasure_geometry() {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
     }
 
     let Ok(object_size) = usize::try_from(candidate.size) else {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
     };
     if candidate.parts.first().is_none_or(|part| part.size != object_size) {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
     }
     if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
     }
 
     let Ok(erasure) = coding::Erasure::try_new_with_options(
@@ -690,18 +709,18 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_verified(
         candidate.erasure.block_size,
         candidate.uses_legacy_checksum,
     ) else {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
     };
-    let Some(data_files) =
+    let data_files =
-        collect_inline_data_shard_fileinfos_by_index(parts_metadata, candidate, erasure.data_shards, |index| {
+        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
             disks.get(index).is_some_and(Option::is_some)
-        })
+        }) {
-    else {
+            Ok(data_files) => data_files,
-        return false;
+            Err(reason) => return Some(reason),
-    };
+        };
 
     let Some(part) = candidate.parts.first() else {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
     };
     let checksum_info = candidate.erasure.get_checksum_info(part.number);
     let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -721,12 +740,13 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_verified(
     let Ok(mut readers) =
         build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
     else {
-        return false;
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
     };
 
-    try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size)
+    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        .await
+        Some(body) if body.len() == object_size => None,
-        .is_some_and(|body| body.len() == object_size)
+        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+    }
 }
 
 pub(in crate::set_disk) fn classify_metadata_response_error(err: &DiskError) -> &'static str {
@@ -2469,6 +2489,7 @@ impl SetDisks {
         let mut next_fanout_index = 0usize;
         let mut scheduled_count = 0usize;
         let mut force_full_wait = false;
+        let mut final_miss_reason_override = None;
         let spawn_read_version =
             |join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
                 let task_opts = opts;
@@ -2541,17 +2562,29 @@ impl SetDisks {
                     .or_else(|| accumulator.version_early_stop_decision())
             {
                 let should_return_early = if read_data {
-                    let allow_data_read_early_stop = match accumulator.candidate.as_ref() {
+                    match accumulator.candidate.as_ref() {
-                        Some(candidate) => {
+                        Some(candidate) => match data_read_early_stop_inline_body_miss_reason(
-                            data_read_early_stop_inline_body_verified(bucket.as_ref(), object.as_ref(), candidate, &ress, disks)
+                            bucket.as_ref(),
-                                .await
+                            object.as_ref(),
+                            candidate,
+                            &ress,
+                            disks,
+                        )
+                        .await
+                        {
+                            None => true,
+                            Some(reason) => {
+                                force_full_wait = true;
+                                final_miss_reason_override = Some(reason);
+                                false
+                            }
+                        },
+                        None => {
+                            force_full_wait = true;
+                            final_miss_reason_override = Some(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM);
+                            false
                         }
-                        None => false,
-                    };
-                    if !allow_data_read_early_stop {
-                        force_full_wait = true;
                     }
-                    allow_data_read_early_stop
                 } else {
                     true
                 };
@@ -2613,7 +2646,12 @@ impl SetDisks {
             }
         }
 
-        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, accumulator.final_miss_reason());
+        let accumulator_miss_reason = accumulator.final_miss_reason();
+        let final_miss_reason = match (final_miss_reason_override, accumulator_miss_reason) {
+            (Some(reason), GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM) => reason,
+            _ => accumulator_miss_reason,
+        };
+        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, final_miss_reason);
         rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses(metrics_path, 0);
         rustfs_io_metrics::record_get_object_metadata_fanout_lifecycle(metrics_path, scheduled_count, scheduled_count, 0);
         let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
@@ -6067,11 +6105,126 @@ mod tests {
             .clone();
 
         assert!(
-            data_read_early_stop_inline_body_verified(bucket, object, &candidate, &parts_metadata, &disks).await,
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks)
+                .await
+                .is_none(),
             "legacy inline metadata must use the legacy bitrot shard sizing and checksum algorithm"
         );
     }
 
+    #[tokio::test]
+    async fn data_read_early_stop_reports_inline_miss_reasons() {
+        let bucket = "inline-data-get-miss-reason-bucket";
+        let object = "inline-data-get-miss-reason-object";
+        let payload = b"verified inline payload";
+        let (_dirs, disks) = call_counter_local_disks(bucket, 4).await;
+        let files = inline_metadata_fanout_fileinfos_with_mode(bucket, object, payload, false).await;
+        let distribution = files
+            .first()
+            .map(|file| file.erasure.distribution.clone())
+            .expect("fixture should include metadata");
+        let order = bounded_metadata_fanout_order(bucket, object, 4, 2);
+        let mut parts_metadata = vec![FileInfo::default(); 4];
+        for disk_index in order.into_iter().take(3) {
+            let block_index = distribution
+                .get(disk_index)
+                .copied()
+                .expect("fixture distribution should cover every disk");
+            parts_metadata[disk_index] = files
+                .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
+                .expect("fixture should include every distributed shard")
+                .clone();
+        }
+        let candidate = parts_metadata
+            .iter()
+            .find(|file| file.name == object)
+            .expect("fixture should include observed metadata")
+            .clone();
+        let data_disk = distribution
+            .iter()
+            .position(|block_index| *block_index == 1)
+            .expect("fixture distribution should include first data shard");
+
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks).await,
+            None
+        );
+
+        let mut not_inline = candidate.clone();
+        rustfs_utils::http::remove_str(&mut not_inline.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA);
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &not_inline, &parts_metadata, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE)
+        );
+
+        let mut transformed = candidate.clone();
+        rustfs_utils::http::insert_str(&mut transformed.metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &transformed, &parts_metadata, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED)
+        );
+
+        let mut deleted = candidate.clone();
+        deleted.deleted = true;
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &deleted, &parts_metadata, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED)
+        );
+
+        let mut zero_size = candidate.clone();
+        zero_size.size = 0;
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &zero_size, &parts_metadata, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE)
+        );
+
+        let mut multipart = candidate.clone();
+        multipart.parts.push(multipart.parts[0].clone());
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &multipart, &parts_metadata, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE)
+        );
+
+        let mut invalid_geometry = candidate.clone();
+        invalid_geometry.erasure.data_blocks = 0;
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &invalid_geometry, &parts_metadata, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY)
+        );
+
+        let mut missing_shard = parts_metadata.clone();
+        missing_shard[data_disk] = FileInfo::default();
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_shard, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
+        );
+
+        let mut missing_payload = parts_metadata.clone();
+        missing_payload[data_disk].data = None;
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_payload, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD)
+        );
+
+        let mut identity_mismatch = parts_metadata.clone();
+        identity_mismatch[data_disk].version_id = Some(Uuid::new_v4());
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &identity_mismatch, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH)
+        );
+
+        let mut corrupt = parts_metadata.clone();
+        if let Some(data) = corrupt[data_disk].data.as_mut() {
+            let mut corrupt_data = data.to_vec();
+            corrupt_data[0] ^= 0x01;
+            *data = Bytes::from(corrupt_data);
+        }
+        assert_eq!(
+            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &corrupt, &disks).await,
+            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY)
+        );
+    }
+
     #[test]
     #[serial_test::serial]
     fn metadata_fanout_lifecycle_records_real_early_stop_abort() {
@@ -6161,7 +6314,7 @@ mod tests {
                 &[
                     ("path", GET_OBJECT_PATH_INTERNAL_META),
                     ("decision", "miss"),
-                    ("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
+                    ("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
                 ],
             ),
             1,
@@ -6173,7 +6326,7 @@ mod tests {
                 &[
                     ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
                     ("decision", "miss"),
-                    ("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
+                    ("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
                 ],
             ),
             0,

crates/ecstore/src/set_disk/mod.rs

@@ -59,7 +59,10 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
 use crate::cluster::rpc::heal_bucket_local_on_disks;
 use crate::data_usage::record_compression_total_memory;
 use crate::diagnostics::get::{
-    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
+    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
+    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
     GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY,
     GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX,
     GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE,
@@ -3866,8 +3869,17 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
     parts_metadata: &'a [FileInfo],
     fi: &FileInfo,
     data_shards: usize,
-    mut disk_is_online: impl FnMut(usize) -> bool,
+    disk_is_online: impl FnMut(usize) -> bool,
 ) -> Option<Vec<&'a FileInfo>> {
+    collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, fi, data_shards, disk_is_online).ok()
+}
+
+fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
+    parts_metadata: &'a [FileInfo],
+    fi: &FileInfo,
+    data_shards: usize,
+    mut disk_is_online: impl FnMut(usize) -> bool,
+) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
     let distribution = &fi.erasure.distribution;
     let mut data_files = vec![None; data_shards];
 
@@ -3875,27 +3887,35 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
         if !disk_is_online(disk_index) {
             continue;
         }
-        let block_index = *distribution.get(disk_index)?;
+        let Some(&block_index) = distribution.get(disk_index) else {
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+        };
         if block_index == 0 || block_index > data_shards {
             continue;
         }
+        if file_info.name.is_empty() {
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD);
+        }
         if file_info.erasure.index != block_index {
-            continue;
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
         }
         if !file_info.has_valid_erasure_geometry() {
-            continue;
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
         }
         if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
-            continue;
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
         }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
-            continue;
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD);
         }
 
         data_files[block_index - 1] = Some(file_info);
     }
 
-    data_files.into_iter().collect()
+    data_files
+        .into_iter()
+        .collect::<Option<Vec<_>>>()
+        .ok_or(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
 }
 
 impl SetDisks {

crates/filemeta/src/filemeta/codec.rs

@@ -135,10 +135,7 @@ impl FileMeta {
         let i = buf.len() as u64;
 
         // check version, buf = buf[8..]
-        let (buf, _, _) = Self::check_xl2_v1(buf).map_err(|e| {
+        let (buf, _, _) = Self::check_xl2_v1(buf)?;
-            error!("failed to check XL2 v1 format: {}", e);
-            e
-        })?;
 
         if buf.len() < 5 {
             error!(

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/scanner/Cargo.toml

@@ -102,7 +102,7 @@ bytes.workspace = true
 hex-simd.workspace = true
 
 [dev-dependencies]
-tracing-subscriber = { workspace = true, features = ["env-filter", "time"] }
+tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
 serial_test = { workspace = true }
 temp-env = { workspace = true }
 tempfile = { workspace = true }

crates/scanner/src/scanner_folder.rs

@@ -65,6 +65,7 @@ const LOG_SUBSYSTEM_FOLDER: &str = "folder";
 const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
 const LOG_SUBSYSTEM_HEAL: &str = "heal";
 const EVENT_SCANNER_FOLDER_STATE: &str = "scanner_folder_state";
+const EVENT_SCANNER_METADATA_CORRUPT: &str = "scanner_metadata_corrupt";
 const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action";
 const EVENT_SCANNER_HEAL_ADMISSION: &str = "scanner_heal_admission";
 const EVENT_SCANNER_ALERT_STATE: &str = "scanner_alert_state";
@@ -2154,17 +2155,34 @@ impl FolderScanner {
                             self.record_failed(&item.path);
 
                             if should_log_failed_object(into.failed_objects) {
-                                warn!(
+                                if let GetSizeFailureAction::HealMetadata { object } = &failure_action {
-                                    target: "rustfs::scanner::folder",
+                                    error!(
-                                    event = EVENT_SCANNER_FOLDER_STATE,
+                                        target: "rustfs::scanner::folder",
-                                    component = LOG_COMPONENT_SCANNER,
+                                        event = EVENT_SCANNER_METADATA_CORRUPT,
-                                    subsystem = LOG_SUBSYSTEM_FOLDER,
+                                        component = LOG_COMPONENT_SCANNER,
-                                    path = %item.path,
+                                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                                    failed_objects = into.failed_objects,
+                                        drive = %self.local_disk.path().display(),
-                                    state = "get_size_failed",
+                                        bucket = %item.bucket,
-                                    error = %e,
+                                        object = %object,
-                                    "Scanner folder failed to get object size"
+                                        metadata_path = %item.path,
-                                );
+                                        failed_objects = into.failed_objects,
+                                        state = "metadata_corrupt",
+                                        error = %e,
+                                        "Scanner detected corrupt object metadata"
+                                    );
+                                } else {
+                                    warn!(
+                                        target: "rustfs::scanner::folder",
+                                        event = EVENT_SCANNER_FOLDER_STATE,
+                                        component = LOG_COMPONENT_SCANNER,
+                                        subsystem = LOG_SUBSYSTEM_FOLDER,
+                                        path = %item.path,
+                                        failed_objects = into.failed_objects,
+                                        state = "get_size_failed",
+                                        error = %e,
+                                        "Scanner folder failed to get object size"
+                                    );
+                                }
                             }
                         }
 
@@ -3054,12 +3072,59 @@ mod tests {
     use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
     use rustfs_filemeta::{FileInfo, FileMeta};
     use serial_test::serial;
+    use std::io::Write;
     #[cfg(unix)]
     use std::os::unix::fs::{PermissionsExt, symlink};
+    use std::sync::Mutex;
     use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
     use temp_env::{with_var, with_var_unset};
+    use tracing_subscriber::fmt::MakeWriter;
     use uuid::Uuid;
 
+    #[derive(Clone, Default)]
+    struct CapturedLogs {
+        buffer: Arc<Mutex<Vec<u8>>>,
+    }
+
+    struct CapturedLogWriter {
+        buffer: Arc<Mutex<Vec<u8>>>,
+    }
+
+    impl CapturedLogs {
+        fn contents(&self) -> String {
+            let buffer = self
+                .buffer
+                .lock()
+                .expect("captured logs mutex should not be poisoned")
+                .clone();
+            String::from_utf8(buffer).expect("captured logs should be valid UTF-8")
+        }
+    }
+
+    impl Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.buffer
+                .lock()
+                .expect("captured logs mutex should not be poisoned")
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedLogWriter {
+                buffer: Arc::clone(&self.buffer),
+            }
+        }
+    }
+
     #[test]
     fn scanner_size_summary_application_saturates_usage_counters() {
         let target = "arn:minio:replication::target".to_string();
@@ -4542,9 +4607,19 @@ mod tests {
         assert!(budget.entries_visited() >= 1);
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "current_thread")]
     #[serial]
     async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .json()
+            .with_max_level(tracing::Level::ERROR)
+            .with_writer(logs.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+        let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+
         let (mut scanner, temp_dir) = build_test_scanner().await;
         let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
 
@@ -4596,6 +4671,30 @@ mod tests {
         assert!(!budget.budget_elapsed());
         assert_eq!(budget.reason(), None);
 
+        let captured = logs.contents();
+        assert!(
+            !captured.contains("failed to check XL2 v1 format"),
+            "the context-free filemeta parser error must not be emitted"
+        );
+        let events = captured
+            .lines()
+            .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("captured scanner log should be valid JSON"))
+            .filter(|line| line["fields"]["event"] == EVENT_SCANNER_METADATA_CORRUPT)
+            .collect::<Vec<_>>();
+        assert_eq!(
+            events.len(),
+            1,
+            "one corrupt metadata observation must emit one scanner-owned diagnostic event"
+        );
+        let fields = &events[0]["fields"];
+        assert_eq!(fields["component"], LOG_COMPONENT_SCANNER);
+        assert_eq!(fields["subsystem"], LOG_SUBSYSTEM_FOLDER);
+        assert_eq!(fields["drive"], temp_dir.to_string_lossy().as_ref());
+        assert_eq!(fields["bucket"], "bucket");
+        assert_eq!(fields["object"], "object");
+        assert_eq!(fields["metadata_path"], metadata_path.to_string_lossy().as_ref());
+        assert_eq!(fields["state"], "metadata_corrupt");
+
         let retry_budget = ScannerCycleBudget::new_with_progress_tracking(
             &parent,
             crate::scanner_budget::ScannerCycleBudgetConfig {

crates/scanner/src/scanner_io.rs

@@ -3849,17 +3849,6 @@ impl ScannerIODisk for Disk {
         let fivs = match meta.get_file_info_versions(item.bucket.as_str(), item.object_path().as_str(), false) {
             Ok(versions) => versions,
             Err(e) => {
-                error!(
-                    target: "rustfs::scanner::io",
-                    event = EVENT_SCANNER_DISK_BUCKET_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_IO,
-                    bucket = %item.bucket,
-                    object = %item.object_path(),
-                    state = "file_info_versions_failed",
-                    error = %e,
-                    "Scanner disk bucket failed to resolve file info versions"
-                );
                 return Err(scanner_metadata_corrupt_error(
                     format!("failed to resolve file info versions: {e}"),
                     &item.bucket,

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
+    // restarts its generation counter — a mark left over from the previous
-    // previous membership would then reject every edit it sends after it
+    // membership must not reject the edits it sends after it rejoins. This
-    // rejoins. Dropping departed origins on load also keeps the map bounded.
+    // 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));
@@ -5935,11 +5939,51 @@ fn summarize_peer_error_detail(detail: &str) -> String {
     summary
 }
 
-/// Allocate the next peer-edit generation. Called inside the state
+/// The wall clock in unix nanoseconds, clamped into u64. A pre-1970 (or
-/// transaction, so the counter is handed out under the distributed
+/// post-2554) clock yields 0, which makes the hybrid allocation below
-/// state-object lock and two nodes of this site can never take the same one.
+/// 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
 }
 
@@ -13244,6 +13288,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 +16186,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,
+            0,
-            "the persisted counter must account for every allocation"
+            "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"
         );
     }
 

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 }