fix(scanner): hold buckets with in-flight namespace commits instead of vetoing

A single node under continuous writes always has some namespace commit in
flight, so the instance-wide pending-commit count vetoed every post-walk
check as `publication_blocked` (Superseded) and deferred every publication.
The same count also missed an early-ACK tail that ended between the walk and
the check, whose S3 dirty mark predates the walk.

Attribute each namespace commit to its bucket (and object) in a per-instance
card table: in flight from before the first disk effect, stamped with a
monotonic end sequence when the last owner drops. A cycle records the end
sequence as its floor before walking and, after the walk, holds every listed
bucket with a commit in flight or ended above the floor grey: the bucket is
reported pending and left out of certification. Dirty buckets keep the
generation and producer identities their S3 marks recorded (only the scope
widens), so segment-reuse evidence is unchanged; buckets that are not dirty
(heal rewrites have no S3 mark) are shaded so the next cycle walks them.

Only commits that cannot be attributed still veto: a guard without a bucket
name, counter exhaustion, or an end that did not fit the bounded table.
Every production call site, including peer-RPC disk admission, now names its
bucket. No persisted format or activity protocol changes.

Docs: describe the rule under "In-flight namespace commits" in
scanner-per-bucket-convergence.md, update the guardrail and boundedness
tables, and the PUT-tail note in scanner-usage-publication.md.

crates/ecstore/src/api/mod.rs

@@ -609,11 +609,23 @@ pub mod set_disk {
         pub use crate::set_disk::ConditionalReadBenchmarkMetadataGuard;
         pub use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause};
 
-        /// Keep a namespace commit pending until the returned owner is dropped.
+        /// Keep a namespace commit that names no bucket pending until the
+        /// returned owner is dropped.
         #[must_use]
         pub fn hold_namespace_commit(store: &crate::store::ECStore) -> impl Send + Sync {
             store.ctx.begin_namespace_commit()
         }
+
+        /// Keep a namespace commit attributed to `bucket`/`object` pending
+        /// until the returned owner is dropped, as a held PUT rename tail does.
+        #[must_use]
+        pub fn hold_bucket_namespace_commit(
+            store: &crate::store::ECStore,
+            bucket: &str,
+            object: &str,
+        ) -> impl Send + Sync + use<> {
+            store.ctx.begin_bucket_namespace_commit(bucket, Some(object))
+        }
     }
 }
 
@@ -623,6 +635,7 @@ pub mod store_list {
 
 pub mod storage {
     pub use crate::core::pools::{HealLifecycleExpiryContext, POOL_META_NAME};
+    pub use crate::runtime::namespace_commit_cards::{ScannerNamespaceCommitCardScope, ScannerNamespaceCommitCards};
     pub use crate::store::HealWalkVersion;
     pub use crate::store::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,

crates/ecstore/src/runtime/instance.rs

@@ -40,6 +40,7 @@
 //! single-instance behavior is byte-for-byte unchanged.
 
 use super::global::TypeLocalDiskSetDrives;
+use super::namespace_commit_cards::{NamespaceCommitCardTable, NamespaceCommitKey, ScannerNamespaceCommitCards};
 use crate::bucket::bandwidth::monitor::Monitor;
 use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState};
 use crate::bucket::metadata_sys::BucketMetadataSys;
@@ -94,10 +95,36 @@ pub(crate) struct ScannerPublicationLeaseEntry {
 pub(crate) struct NamespaceCommitGuard {
     ctx: Arc<InstanceContext>,
     counted: bool,
+    card: NamespaceCommitCard,
+}
+
+/// How a namespace commit is attributed for the scanner.
+enum NamespaceCommitCard {
+    /// The owner cannot name its bucket.
+    Unscoped,
+    Bucket(NamespaceCommitKey),
+    /// Meta-bucket commits are counted but never carded: the scanner does not
+    /// walk meta buckets for usage.
+    Untracked,
+}
+
+impl NamespaceCommitCard {
+    fn key(&self) -> Option<Option<&NamespaceCommitKey>> {
+        match self {
+            Self::Unscoped => Some(None),
+            Self::Bucket(key) => Some(Some(key)),
+            Self::Untracked => None,
+        }
+    }
 }
 
 impl Drop for NamespaceCommitGuard {
     fn drop(&mut self) {
+        // The last owner drops after the last disk task, so the ended card is
+        // recorded strictly after every disk effect of this commit.
+        if let Some(key) = self.card.key() {
+            self.ctx.namespace_commit_cards().end(key);
+        }
         if self.counted {
             // Publish the new generation before a zero-pending publication probe.
             self.ctx.advance_namespace_commit_generation();
@@ -239,6 +266,10 @@ pub struct InstanceContext {
     scanner_publication_state: AtomicU8,
     namespace_commits: AtomicU64,
     namespace_commit_generation: AtomicU64,
+    /// Per-bucket attribution of the commits counted by `namespace_commits`.
+    /// Leaf lock: held only for in-memory map updates, never across another
+    /// lock or an await.
+    namespace_commit_cards: std::sync::Mutex<NamespaceCommitCardTable>,
     /// Resolves object-encryption material at the application boundary.
     object_encryption_resolver: OnceLock<Arc<dyn ObjectEncryptionResolver>>,
     delete_recovery_hints: DeleteRecoveryHints,
@@ -295,6 +326,7 @@ impl InstanceContext {
             scanner_publication_state: AtomicU8::new(SCANNER_PUBLICATION_STATE_UNKNOWN),
             namespace_commits: AtomicU64::new(0),
             namespace_commit_generation: AtomicU64::new(0),
+            namespace_commit_cards: std::sync::Mutex::new(NamespaceCommitCardTable::default()),
             object_encryption_resolver: OnceLock::new(),
             delete_recovery_hints: DeleteRecoveryHints::default(),
             tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()),
@@ -442,7 +474,37 @@ impl InstanceContext {
             && self.scanner_publication_state.load(Ordering::Acquire) == SCANNER_PUBLICATION_STATE_ALLOWED
     }
 
+    /// Begin a namespace commit that cannot name its bucket. While it is in
+    /// flight the scanner treats every walk as unattributable and keeps the
+    /// whole-cycle publication barrier. Production commits name their bucket
+    /// through `begin_bucket_namespace_commit`.
+    #[cfg(any(test, feature = "test-util"))]
     pub(crate) fn begin_namespace_commit(self: &Arc<Self>) -> Arc<NamespaceCommitGuard> {
+        self.begin_namespace_commit_with_card(NamespaceCommitCard::Unscoped)
+    }
+
+    /// Begin a namespace commit on `bucket` (and `object`, when the commit
+    /// changes one object). The scanner shades that bucket while the commit is
+    /// in flight and after it ends, instead of vetoing the whole cycle.
+    pub(crate) fn begin_bucket_namespace_commit(
+        self: &Arc<Self>,
+        bucket: &str,
+        object: Option<&str>,
+    ) -> Arc<NamespaceCommitGuard> {
+        let card = if bucket.is_empty() {
+            NamespaceCommitCard::Unscoped
+        } else if crate::bucket::utils::is_meta_bucketname(bucket) {
+            NamespaceCommitCard::Untracked
+        } else {
+            NamespaceCommitCard::Bucket(NamespaceCommitKey::new(bucket, object))
+        };
+        self.begin_namespace_commit_with_card(card)
+    }
+
+    fn begin_namespace_commit_with_card(self: &Arc<Self>, card: NamespaceCommitCard) -> Arc<NamespaceCommitGuard> {
+        if let Some(key) = card.key() {
+            self.namespace_commit_cards().begin(key);
+        }
         let counted = self
             .namespace_commits
             .try_update(Ordering::AcqRel, Ordering::Acquire, |count| count.checked_add(1))
@@ -455,9 +517,35 @@ impl InstanceContext {
         Arc::new(NamespaceCommitGuard {
             ctx: Arc::clone(self),
             counted,
+            card,
         })
     }
 
+    fn namespace_commit_cards(&self) -> std::sync::MutexGuard<'_, NamespaceCommitCardTable> {
+        self.namespace_commit_cards
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    /// Whether an in-flight namespace commit cannot be attributed to a bucket:
+    /// an unscoped owner, or an exhausted commit counter.
+    pub(crate) fn namespace_commits_unattributed(&self) -> bool {
+        self.namespace_commit_cards().unscoped_in_flight() || self.namespace_commit_generation() == u64::MAX
+    }
+
+    /// End sequence a scanner walk records before it starts reading.
+    pub(crate) fn namespace_commit_card_floor(&self) -> u64 {
+        self.namespace_commit_cards().floor()
+    }
+
+    /// Buckets whose namespace commits are in flight or ended after `floor`,
+    /// and whether any such commit cannot be attributed to a bucket.
+    pub(crate) fn namespace_commit_cards_overlapping(&self, floor: u64) -> ScannerNamespaceCommitCards {
+        let mut cards = self.namespace_commit_cards().overlapping(floor);
+        cards.unattributed |= self.namespace_commit_generation() == u64::MAX;
+        cards
+    }
+
     fn advance_namespace_commit_generation(&self) {
         let _ = self
             .namespace_commit_generation
@@ -905,6 +993,56 @@ mod tests {
         }
     }
 
+    #[test]
+    fn bucket_namespace_commits_are_carded_and_only_unscoped_or_exhausted_ones_are_unattributed() {
+        use crate::runtime::namespace_commit_cards::ScannerNamespaceCommitCardScope;
+        use std::collections::{BTreeMap, BTreeSet};
+
+        let ctx = Arc::new(InstanceContext::new());
+        let floor = ctx.namespace_commit_card_floor();
+        let tail = ctx.begin_bucket_namespace_commit("photos", Some("2026/a.jpg"));
+        let shared_tail = Arc::clone(&tail);
+        let meta = ctx.begin_bucket_namespace_commit(crate::disk::RUSTFS_META_BUCKET, Some("config.json"));
+        assert!(ctx.namespace_commits_pending(), "attributed commits still count as pending");
+        assert!(!ctx.namespace_commits_unattributed());
+        let photos = BTreeMap::from([(
+            "photos".to_string(),
+            ScannerNamespaceCommitCardScope::Objects(BTreeSet::from(["2026/a.jpg".to_string()])),
+        )]);
+        let in_flight = ctx.namespace_commit_cards_overlapping(floor);
+        assert!(!in_flight.unattributed);
+        assert_eq!(in_flight.buckets, photos, "meta-bucket commits are counted but never carded");
+
+        drop(tail);
+        assert_eq!(ctx.namespace_commit_card_floor(), floor, "an end is recorded by the last owner only");
+        drop(shared_tail);
+        drop(meta);
+        assert!(!ctx.namespace_commits_pending());
+        assert_eq!(ctx.namespace_commit_cards_overlapping(floor).buckets, photos);
+        assert!(
+            ctx.namespace_commit_cards_overlapping(ctx.namespace_commit_card_floor())
+                .buckets
+                .is_empty()
+        );
+
+        let unscoped = ctx.begin_namespace_commit();
+        assert!(ctx.namespace_commits_unattributed());
+        assert!(
+            ctx.namespace_commit_cards_overlapping(ctx.namespace_commit_card_floor())
+                .unattributed
+        );
+        drop(unscoped);
+        assert!(!ctx.namespace_commits_unattributed());
+
+        ctx.namespace_commit_generation.store(u64::MAX - 1, Ordering::Release);
+        drop(ctx.begin_bucket_namespace_commit("photos", None));
+        assert!(ctx.namespace_commits_unattributed(), "an exhausted counter cannot be attributed");
+        assert!(
+            ctx.namespace_commit_cards_overlapping(ctx.namespace_commit_card_floor())
+                .unattributed
+        );
+    }
+
     #[tokio::test(start_paused = true)]
     async fn scanner_lease_generations_remain_bound_until_expiry() {
         let ctx = Arc::new(InstanceContext::new());

crates/ecstore/src/runtime/mod.rs

@@ -16,4 +16,5 @@
 
 pub(crate) mod global;
 pub(crate) mod instance;
+pub(crate) mod namespace_commit_cards;
 pub(crate) mod sources;

crates/ecstore/src/runtime/namespace_commit_cards.rs

@@ -0,0 +1,320 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Per-bucket attribution of namespace commits for the scanner.
+//!
+//! A namespace commit is registered here under its bucket (and object, when it
+//! has one) before its first disk effect. When its last owner drops, after its
+//! last disk task, the bucket is stamped with a monotonic end sequence. The
+//! scanner reads the sequence before a walk and, after the walk, asks for every
+//! bucket that is still in flight or ended past that floor: those buckets may
+//! have been walked mid-commit and stay grey instead of vetoing the cycle.
+//! Commits that cannot be attributed (unscoped owners, or an ended card that
+//! did not fit the table) are reported as such and keep the whole-cycle
+//! barrier.
+
+use std::collections::{BTreeMap, BTreeSet, HashMap};
+
+/// Distinct buckets the ended table may hold. An end that does not fit is
+/// reported as unattributed to any walk that overlaps it.
+pub(crate) const MAX_NAMESPACE_COMMIT_CARD_BUCKETS: usize = 65_536;
+
+/// Objects kept per bucket before the card collapses to the whole bucket. It
+/// matches the scanner's top-level-entry bound for one dirty bucket.
+pub(crate) const MAX_NAMESPACE_COMMIT_CARD_OBJECTS_PER_BUCKET: usize = 128;
+
+/// Object names kept across all ended cards. Past it, a bucket that would add
+/// another name collapses to the whole bucket, so a scanner that stops
+/// draining cannot grow the table beyond its bucket bound.
+pub(crate) const MAX_NAMESPACE_COMMIT_CARD_OBJECTS: usize = 65_536;
+
+/// The part of a bucket that a namespace commit may have changed.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ScannerNamespaceCommitCardScope {
+    WholeBucket,
+    Objects(BTreeSet<String>),
+}
+
+/// Buckets whose namespace commits overlap a scanner walk.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct ScannerNamespaceCommitCards {
+    pub buckets: BTreeMap<String, ScannerNamespaceCommitCardScope>,
+    /// A commit overlapping the walk could not be attributed to a bucket.
+    pub unattributed: bool,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub(crate) struct NamespaceCommitKey {
+    bucket: String,
+    object: Option<String>,
+}
+
+impl NamespaceCommitKey {
+    pub(crate) fn new(bucket: &str, object: Option<&str>) -> Self {
+        Self {
+            bucket: bucket.to_string(),
+            object: object.filter(|object| !object.is_empty()).map(str::to_string),
+        }
+    }
+}
+
+#[derive(Debug)]
+struct EndedCard {
+    last_end: u64,
+    scope: ScannerNamespaceCommitCardScope,
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct NamespaceCommitCardTable {
+    end_sequence: u64,
+    in_flight: HashMap<NamespaceCommitKey, usize>,
+    unscoped: usize,
+    ended: BTreeMap<String, EndedCard>,
+    /// Object names held by `ended`.
+    ended_objects: usize,
+    /// End sequence of the latest end that did not fit `ended`.
+    overflow_end: u64,
+}
+
+fn scope_objects(scope: &ScannerNamespaceCommitCardScope) -> usize {
+    match scope {
+        ScannerNamespaceCommitCardScope::WholeBucket => 0,
+        ScannerNamespaceCommitCardScope::Objects(objects) => objects.len(),
+    }
+}
+
+fn merge_scope(scope: &mut ScannerNamespaceCommitCardScope, object: Option<&String>) {
+    match (&mut *scope, object) {
+        (ScannerNamespaceCommitCardScope::WholeBucket, _) => {}
+        (ScannerNamespaceCommitCardScope::Objects(_), None) => *scope = ScannerNamespaceCommitCardScope::WholeBucket,
+        (ScannerNamespaceCommitCardScope::Objects(objects), Some(object)) => {
+            objects.insert(object.clone());
+            if objects.len() > MAX_NAMESPACE_COMMIT_CARD_OBJECTS_PER_BUCKET {
+                *scope = ScannerNamespaceCommitCardScope::WholeBucket;
+            }
+        }
+    }
+}
+
+fn empty_scope() -> ScannerNamespaceCommitCardScope {
+    ScannerNamespaceCommitCardScope::Objects(BTreeSet::new())
+}
+
+impl NamespaceCommitCardTable {
+    pub(crate) fn begin(&mut self, key: Option<&NamespaceCommitKey>) {
+        match key {
+            Some(key) => *self.in_flight.entry(key.clone()).or_default() += 1,
+            None => self.unscoped = self.unscoped.saturating_add(1),
+        }
+    }
+
+    pub(crate) fn end(&mut self, key: Option<&NamespaceCommitKey>) {
+        let Some(key) = key else {
+            self.unscoped = self.unscoped.saturating_sub(1);
+            return;
+        };
+        if let Some(count) = self.in_flight.get_mut(key) {
+            *count -= 1;
+            if *count == 0 {
+                self.in_flight.remove(key);
+            }
+        }
+        // Saturation only merges later ends into the same sequence; a floor
+        // read at `u64::MAX` then reports every card, which is conservative.
+        self.end_sequence = self.end_sequence.saturating_add(1);
+        let end = self.end_sequence;
+        if !self.ended.contains_key(&key.bucket) && self.ended.len() >= MAX_NAMESPACE_COMMIT_CARD_BUCKETS {
+            self.overflow_end = end;
+            return;
+        }
+        let card = self.ended.entry(key.bucket.clone()).or_insert_with(|| EndedCard {
+            last_end: end,
+            scope: empty_scope(),
+        });
+        card.last_end = end;
+        let before = scope_objects(&card.scope);
+        if self.ended_objects >= MAX_NAMESPACE_COMMIT_CARD_OBJECTS
+            && key.object.as_ref().is_some_and(
+                |object| matches!(&card.scope, ScannerNamespaceCommitCardScope::Objects(objects) if !objects.contains(object)),
+            )
+        {
+            card.scope = ScannerNamespaceCommitCardScope::WholeBucket;
+        } else {
+            merge_scope(&mut card.scope, key.object.as_ref());
+        }
+        self.ended_objects = self.ended_objects - before + scope_objects(&card.scope);
+    }
+
+    /// Sequence a walk records before it starts reading.
+    pub(crate) fn floor(&self) -> u64 {
+        self.end_sequence
+    }
+
+    /// Whether an in-flight commit cannot be attributed to a bucket.
+    pub(crate) fn unscoped_in_flight(&self) -> bool {
+        self.unscoped != 0
+    }
+
+    /// Every bucket with a commit in flight or ended after `floor`. Ended cards
+    /// at or below `floor` ended before the walk that recorded it started and
+    /// are pruned: one collector runs per instance and its floors only grow.
+    pub(crate) fn overlapping(&mut self, floor: u64) -> ScannerNamespaceCommitCards {
+        self.ended.retain(|_, card| card.last_end > floor || floor == u64::MAX);
+        self.ended_objects = self.ended.values().map(|card| scope_objects(&card.scope)).sum();
+        let mut buckets = self
+            .ended
+            .iter()
+            .map(|(bucket, card)| (bucket.clone(), card.scope.clone()))
+            .collect::<BTreeMap<_, _>>();
+        for key in self.in_flight.keys() {
+            merge_scope(buckets.entry(key.bucket.clone()).or_insert_with(empty_scope), key.object.as_ref());
+        }
+        for scope in buckets.values_mut() {
+            if matches!(scope, ScannerNamespaceCommitCardScope::Objects(objects) if objects.is_empty()) {
+                *scope = ScannerNamespaceCommitCardScope::WholeBucket;
+            }
+        }
+        ScannerNamespaceCommitCards {
+            buckets,
+            unattributed: self.unscoped != 0 || self.overflow_end > floor || floor == u64::MAX,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn objects(names: &[&str]) -> ScannerNamespaceCommitCardScope {
+        ScannerNamespaceCommitCardScope::Objects(names.iter().map(|name| name.to_string()).collect())
+    }
+
+    #[test]
+    fn a_walk_sees_commits_in_flight_or_ended_after_its_floor() {
+        let mut table = NamespaceCommitCardTable::default();
+        let before = NamespaceCommitKey::new("before", Some("a"));
+        table.begin(Some(&before));
+        table.end(Some(&before));
+        let hot = NamespaceCommitKey::new("hot", Some("a/1"));
+        table.begin(Some(&hot));
+        table.begin(Some(&hot));
+
+        let floor = table.floor();
+        let during = NamespaceCommitKey::new("during", Some("b/2"));
+        table.begin(Some(&during));
+        table.end(Some(&during));
+        table.end(Some(&hot));
+
+        let overlapping = table.overlapping(floor);
+        assert!(!overlapping.unattributed);
+        assert_eq!(
+            overlapping.buckets,
+            BTreeMap::from([
+                ("during".to_string(), objects(&["b/2"])),
+                ("hot".to_string(), objects(&["a/1"])),
+            ]),
+            "a commit that ended before the floor is not reported"
+        );
+
+        let next_floor = table.floor();
+        assert_eq!(
+            table.overlapping(next_floor).buckets,
+            BTreeMap::from([("hot".to_string(), objects(&["a/1"]))]),
+            "the second owner keeps the commit in flight"
+        );
+        table.end(Some(&hot));
+        assert_eq!(table.overlapping(next_floor).buckets.len(), 1, "it ended after the floor");
+        assert!(table.overlapping(table.floor()).buckets.is_empty());
+        assert!(table.ended.is_empty(), "cards below the latest floor are pruned");
+    }
+
+    #[test]
+    fn bucket_level_commits_and_object_overflow_collapse_to_the_whole_bucket() {
+        let mut table = NamespaceCommitCardTable::default();
+        for index in 0..=MAX_NAMESPACE_COMMIT_CARD_OBJECTS_PER_BUCKET {
+            let key = NamespaceCommitKey::new("wide", Some(&format!("object-{index}")));
+            table.begin(Some(&key));
+            table.end(Some(&key));
+        }
+        let bucket_heal = NamespaceCommitKey::new("healed", None);
+        table.begin(Some(&bucket_heal));
+        table.end(Some(&bucket_heal));
+        let empty_object = NamespaceCommitKey::new("empty", Some(""));
+        table.begin(Some(&empty_object));
+
+        assert_eq!(
+            table.overlapping(0).buckets,
+            BTreeMap::from([
+                ("empty".to_string(), ScannerNamespaceCommitCardScope::WholeBucket),
+                ("healed".to_string(), ScannerNamespaceCommitCardScope::WholeBucket),
+                ("wide".to_string(), ScannerNamespaceCommitCardScope::WholeBucket),
+            ])
+        );
+    }
+
+    #[test]
+    fn object_names_are_bounded_across_the_table() {
+        let mut table = NamespaceCommitCardTable::default();
+        let buckets = MAX_NAMESPACE_COMMIT_CARD_OBJECTS / MAX_NAMESPACE_COMMIT_CARD_OBJECTS_PER_BUCKET;
+        for bucket in 0..buckets {
+            for object in 0..MAX_NAMESPACE_COMMIT_CARD_OBJECTS_PER_BUCKET {
+                let key = NamespaceCommitKey::new(&format!("bucket-{bucket}"), Some(&format!("object-{object}")));
+                table.begin(Some(&key));
+                table.end(Some(&key));
+            }
+        }
+        assert_eq!(table.ended_objects, MAX_NAMESPACE_COMMIT_CARD_OBJECTS);
+        let repeat = NamespaceCommitKey::new("bucket-0", Some("object-0"));
+        table.begin(Some(&repeat));
+        table.end(Some(&repeat));
+        assert_eq!(table.ended_objects, MAX_NAMESPACE_COMMIT_CARD_OBJECTS, "a known name adds nothing");
+        let extra = NamespaceCommitKey::new("extra", Some("object"));
+        table.begin(Some(&extra));
+        table.end(Some(&extra));
+        let wider = NamespaceCommitKey::new("bucket-1", Some("new"));
+        table.begin(Some(&wider));
+        table.end(Some(&wider));
+        assert_eq!(
+            table.ended_objects,
+            MAX_NAMESPACE_COMMIT_CARD_OBJECTS - MAX_NAMESPACE_COMMIT_CARD_OBJECTS_PER_BUCKET
+        );
+        let cards = table.overlapping(0);
+        assert_eq!(cards.buckets["extra"], ScannerNamespaceCommitCardScope::WholeBucket);
+        assert_eq!(cards.buckets["bucket-1"], ScannerNamespaceCommitCardScope::WholeBucket);
+        assert!(!cards.unattributed, "collapsing a scope loses no attribution");
+    }
+
+    #[test]
+    fn unscoped_owners_and_ended_overflow_are_unattributed() {
+        let mut table = NamespaceCommitCardTable::default();
+        table.begin(None);
+        assert!(table.unscoped_in_flight());
+        assert!(table.overlapping(table.floor()).unattributed);
+        table.end(None);
+        assert!(!table.unscoped_in_flight());
+        assert!(!table.overlapping(table.floor()).unattributed);
+
+        for index in 0..=MAX_NAMESPACE_COMMIT_CARD_BUCKETS {
+            let key = NamespaceCommitKey::new(&format!("bucket-{index}"), None);
+            table.begin(Some(&key));
+            table.end(Some(&key));
+        }
+        let overlapping = table.overlapping(0);
+        assert!(overlapping.unattributed, "a dropped ended card cannot be attributed");
+        assert_eq!(overlapping.buckets.len(), MAX_NAMESPACE_COMMIT_CARD_BUCKETS);
+        let after = table.overlapping(table.floor());
+        assert!(!after.unattributed, "an overflow before the floor does not overlap the walk");
+        assert!(after.buckets.is_empty());
+    }
+}

crates/ecstore/src/set_disk/ops/delete_recovery.rs

@@ -299,7 +299,7 @@ impl SetDisks {
         mut context: DeleteRecoveryContext<'_>,
     ) -> Result<()> {
         if context.namespace_owner.is_none() {
-            context.namespace_owner = Some(self.ctx.begin_namespace_commit());
+            context.namespace_owner = Some(self.ctx.begin_bucket_namespace_commit(bucket, Some(object)));
         }
         let can_commit = context.can_commit;
         let states = inspect(disks, bucket, object, transaction).await;
@@ -427,7 +427,7 @@ impl SetDisks {
             let namespace_owner = context
                 .namespace_owner
                 .clone()
-                .unwrap_or_else(|| self.ctx.begin_namespace_commit());
+                .unwrap_or_else(|| self.ctx.begin_bucket_namespace_commit(bucket, Some(object)));
             context.namespace_owner = Some(Arc::new((namespace_owner, guard.clone())));
         }
         let can_commit = context.can_commit;
@@ -583,7 +583,7 @@ impl SetDisks {
         let namespace_owner = if dry_run {
             None
         } else {
-            Some(namespace_owner.unwrap_or_else(|| self.ctx.begin_namespace_commit()))
+            Some(namespace_owner.unwrap_or_else(|| self.ctx.begin_bucket_namespace_commit(bucket, Some(object))))
         };
         if !dry_run && metadata_may_change {
             self.invalidate_get_object_metadata_cache(bucket, object).await;

crates/ecstore/src/set_disk/ops/heal.rs

@@ -1975,7 +1975,7 @@ impl SetDisks {
                                                     _write_lock_guard.clone(),
                                                     _read_repair_write_lock_guard.clone(),
                                                     bucket_heal_scope.clone(),
-                                                    self.ctx.begin_namespace_commit(),
+                                                    self.ctx.begin_bucket_namespace_commit(bucket, Some(object)),
                                                 ))
                                             });
                                             match disk
@@ -3576,8 +3576,11 @@ impl SetDisks {
                         explicit_delete,
                         repair_marker_source: true,
                         namespace_owner: (!opts.dry_run).then(|| {
-                            Arc::new((scope.clone(), self.ctx.begin_namespace_commit(), _write_lock_guard.clone()))
-                                as Arc<dyn Send + Sync>
+                            Arc::new((
+                                scope.clone(),
+                                self.ctx.begin_bucket_namespace_commit(bucket, Some(object)),
+                                _write_lock_guard.clone(),
+                            )) as Arc<dyn Send + Sync>
                         }),
                         ..crate::set_disk::DeleteRecoveryContext::new(Some(scope.incarnation), &|| {
                             scope.check().is_ok() && !_write_lock_guard.as_deref().is_some_and(|guard| guard.is_lock_lost())

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -3733,10 +3733,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 (&commit_bucket, &commit_object),
                 commit_allows_early_ack,
                 crate::set_disk::core::io_primitives::RenameDataFenceOptions::new(write_quorum, None)
-                    .with_namespace_commit_guard(
-                        (!crate::bucket::utils::is_meta_bucketname(&commit_bucket))
-                            .then(|| commit_set.ctx.begin_namespace_commit()),
-                    ),
+                    .with_namespace_commit_guard((!crate::bucket::utils::is_meta_bucketname(&commit_bucket)).then(|| {
+                        commit_set
+                            .ctx
+                            .begin_bucket_namespace_commit(&commit_bucket, Some(&commit_object))
+                    })),
             )
             .await;
             if let Ok(rename_commit) = rename_result.as_mut() {

crates/ecstore/src/set_disk/ops/object.rs

@@ -3848,7 +3848,7 @@ impl SetDisks {
                 DeleteRecoveryContext {
                     explicit_delete: true,
                     namespace_owner: Some(Arc::new((
-                        self.ctx.begin_namespace_commit(),
+                        self.ctx.begin_bucket_namespace_commit(bucket, Some(object)),
                         lock_guard.cloned(),
                         opts.object_lock_config_snapshot.clone(),
                         opts.scanner_publication_commit_scope.clone(),
@@ -5475,9 +5475,11 @@ impl SetDisks {
                     )
                     .with_publication_scope(commit_scanner_publication_scope.clone())
                     .with_rollback_receipt(commit_rollback_receipt.clone())
-                    .with_namespace_commit_guard(
-                        (!is_meta_bucketname(&commit_bucket)).then(|| commit_set.ctx.begin_namespace_commit()),
-                    ),
+                    .with_namespace_commit_guard((!is_meta_bucketname(&commit_bucket)).then(|| {
+                        commit_set
+                            .ctx
+                            .begin_bucket_namespace_commit(&commit_bucket, Some(&commit_object))
+                    })),
                 )
                 .await;
                 rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -8787,7 +8789,7 @@ impl SetDisks {
         let disks = self.disk_inventory().await;
         let namespace_owner = (!is_meta_bucketname(bucket)).then(|| {
             Arc::new((
-                self.ctx.begin_namespace_commit(),
+                self.ctx.begin_bucket_namespace_commit(bucket, Some(object)),
                 commit_context.guard.clone(),
                 commit_context
                     .options
@@ -9890,7 +9892,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         let rollback_dir = Uuid::new_v4();
         let namespace_owner = (!is_meta_bucketname(bucket)).then(|| {
             Arc::new((
-                self.ctx.begin_namespace_commit(),
+                self.ctx.begin_bucket_namespace_commit(bucket, None),
                 opts.delete_mutation_owner.clone(),
                 opts.object_lock_config_snapshot.clone(),
                 opts.scanner_publication_commit_scope.clone(),
@@ -10338,7 +10340,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                     DeleteRecoveryContext {
                         explicit_delete,
                         namespace_owner: Some(Arc::new((
-                            self.ctx.begin_namespace_commit(),
+                            self.ctx.begin_bucket_namespace_commit(bucket, Some(&object.object_name)),
                             opts.delete_mutation_owner.clone(),
                             opts.object_lock_config_snapshot.clone(),
                         ))),
@@ -10650,7 +10652,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                     object,
                     DeleteRecoveryContext {
                         namespace_owner: Some(Arc::new((
-                            self.ctx.begin_namespace_commit(),
+                            self.ctx.begin_bucket_namespace_commit(bucket, Some(object)),
                             _lock_guard.clone(),
                             opts.object_lock_config_snapshot.clone(),
                             opts.scanner_publication_commit_scope.clone(),

crates/ecstore/src/store/heal.rs

@@ -382,7 +382,7 @@ impl ECStore {
                 dry_run: opts.dry_run,
                 explicit_delete,
                 namespace_owner: (!opts.dry_run)
-                    .then(|| Arc::new((scope.clone(), store.ctx.begin_namespace_commit())) as Arc<dyn Send + Sync>),
+                    .then(|| Arc::new((scope.clone(), store.ctx.begin_bucket_namespace_commit(&bucket, None))) as Arc<dyn Send + Sync>),
                 ..crate::set_disk::DeleteRecoveryContext::new(Some(scope.incarnation), &can_commit)
             };
             let mut summary = crate::set_disk::DeleteRecoverySummary::default();

crates/ecstore/src/store/mod.rs

@@ -936,6 +936,33 @@ impl ECStore {
         self.scanner_data_usage_publication_snapshot_blocked().await || self.ctx.namespace_commits_pending()
     }
 
+    /// Local publication barrier for an authoritative scanner snapshot. Data
+    /// movement blocks it as in `scanner_data_usage_publication_blocked`, but
+    /// an in-flight namespace commit blocks it only when it cannot be
+    /// attributed to a bucket: attributed commits are reported by
+    /// `scanner_namespace_commit_cards` and keep their bucket grey instead.
+    pub async fn scanner_attributed_usage_publication_blocked(&self) -> bool {
+        let operation_gate = self.ctx.data_movement_operation_gate();
+        let _operation_guard = operation_gate.read_owned().await;
+        self.scanner_data_usage_publication_snapshot_blocked().await || self.ctx.namespace_commits_unattributed()
+    }
+
+    /// End sequence of namespace commits, recorded by a scanner walk before it
+    /// starts reading. See "In-flight namespace commits" in
+    /// docs/architecture/scanner-per-bucket-convergence.md.
+    pub fn scanner_namespace_commit_card_floor(&self) -> u64 {
+        self.ctx.namespace_commit_card_floor()
+    }
+
+    /// Buckets whose namespace commits were in flight or ended after `floor`,
+    /// that is, commits that may overlap the walk that recorded `floor`.
+    pub fn scanner_namespace_commit_cards(
+        &self,
+        floor: u64,
+    ) -> crate::runtime::namespace_commit_cards::ScannerNamespaceCommitCards {
+        self.ctx.namespace_commit_cards_overlapping(floor)
+    }
+
     pub async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus {
         let operation_gate = self.ctx.data_movement_operation_gate();
         let _operation_guard = operation_gate.read_owned().await;

crates/ecstore/src/store/peer.rs

@@ -88,7 +88,7 @@ impl ECStore {
     /// Resolve a quota lease against this request's captured storage instance.
     pub async fn quota_mutation_fence_disk(&self, disk_ref: &str) -> DiskResult<DiskStore> {
         let (disk, id) = local_disk_candidate(&self.ctx, disk_ref).await?;
-        let _owner = admit_local_disk(&self.ctx, &disk, id, false).await?;
+        let _owner = admit_local_disk(&self.ctx, &disk, id, None).await?;
         Ok(disk)
     }
 
@@ -103,7 +103,7 @@ impl ECStore {
         let path = target.1.to_owned();
         self.run_bucket_heal_at_incarnation(target.0, expected, &Default::default(), move |store, volume, _| async move {
             let (disk, id) = local_disk_candidate(&store.ctx, &disk_ref).await?;
-            let _owner = admit_local_disk(&store.ctx, &disk, id, true).await?;
+            let _owner = admit_local_disk(&store.ctx, &disk, id, Some((&volume, &path))).await?;
             disk.write_metadata("", &volume, &path, fi).await.map_err(Into::into)
         })
         .await
@@ -121,7 +121,7 @@ impl ECStore {
         let path = target.1.to_owned();
         self.run_bucket_heal_at_incarnation(target.0, expected, &Default::default(), move |store, volume, _| async move {
             let (disk, id) = local_disk_candidate(&store.ctx, &disk_ref).await?;
-            let owner = admit_local_disk(&store.ctx, &disk, id, true).await?;
+            let owner = admit_local_disk(&store.ctx, &disk, id, Some((&volume, &path))).await?;
             let scope = super::bucket_heal_scope(&volume).ok_or_else(|| StorageError::other("missing bucket heal scope"))?;
             scope.check()?;
             disk.delete_with_namespace_owner(&volume, &path, opts, Some(Arc::new((scope, owner))))
@@ -150,7 +150,7 @@ impl ECStore {
                 .map_err(|error| DiskError::other(error.to_string()))?,
         );
         let (disk, id) = local_disk_candidate(&self.ctx, disk_ref).await?;
-        let owner = admit_local_disk(&self.ctx, &disk, id, true).await?;
+        let owner = admit_local_disk(&self.ctx, &disk, id, Some(target)).await?;
         if fence.is_lock_lost() {
             return Err(DiskError::other("bucket heal incarnation fence was lost"));
         }
@@ -309,7 +309,7 @@ async fn admit_local_disk(
     ctx: &Arc<InstanceContext>,
     disk: &DiskStore,
     disk_id: Option<Uuid>,
-    mutates_namespace: bool,
+    namespace_target: Option<(&str, &str)>,
 ) -> DiskResult<Option<Arc<NamespaceCommitGuard>>> {
     if !matches!(disk.as_ref(), Disk::Local(_)) {
         return Err(DiskError::DiskNotFound);
@@ -332,7 +332,7 @@ async fn admit_local_disk(
     }
     // Admission linearizes under the registry read: replacement/quarantine
     // before this point rejects; later changes do not revoke physical I/O.
-    Ok(mutates_namespace.then(|| ctx.begin_namespace_commit()))
+    Ok(namespace_target.map(|(bucket, object)| ctx.begin_bucket_namespace_commit(bucket, Some(object))))
 }
 
 async fn rename_local_data_with_ctx(
@@ -350,9 +350,13 @@ async fn rename_local_data_with_ctx(
     if let Some(duration_ms) = lookup_ms {
         rustfs_io_metrics::record_put_object_stage_duration(PUT_STAGE_SET_DISK_RENAME_REMOTE_HANDLER_LOOKUP, duration_ms);
     }
-    let mutates_namespace = !is_meta_bucketname(source.0) || !is_meta_bucketname(destination.0);
+    // Attribute the commit to whichever side is a user bucket; the
+    // destination wins when both are.
+    let namespace_target = [destination, source]
+        .into_iter()
+        .find(|(volume, _)| !is_meta_bucketname(volume));
     let admit_started = rustfs_io_metrics::put_stage_timer();
-    let owner = admit_local_disk(ctx, &disk, disk_id, mutates_namespace).await?;
+    let owner = admit_local_disk(ctx, &disk, disk_id, namespace_target).await?;
     let admit_ms = admit_started.map(|started| started.elapsed().as_secs_f64() * 1000.0);
     if let Some(duration_ms) = admit_ms {
         rustfs_io_metrics::record_put_object_stage_duration(PUT_STAGE_SET_DISK_RENAME_REMOTE_HANDLER_ADMIT, duration_ms);
@@ -390,7 +394,7 @@ async fn undo_local_write_with_ctx(
         return Err(DiskError::other("target undo requires undo_write"));
     }
     let (disk, disk_id) = local_disk_candidate(ctx, disk_ref).await?;
-    let owner = admit_local_disk(ctx, &disk, disk_id, !is_meta_bucketname(volume)).await?;
+    let owner = admit_local_disk(ctx, &disk, disk_id, (!is_meta_bucketname(volume)).then_some((volume, path))).await?;
     let physical_owner = owner.as_ref().map(|owner| owner.clone() as Arc<dyn Send + Sync>);
     let result = disk
         .undo_write_with_namespace_owner(volume, path, fi, opts, physical_owner)
@@ -808,7 +812,10 @@ mod tests {
             }
             drop(entries);
             assert!(
-                matches!(admit_local_disk(&ctx, &disk, None, true).await, Err(DiskError::DiskNotFound)),
+                matches!(
+                    admit_local_disk(&ctx, &disk, None, Some(("bucket", "object"))).await,
+                    Err(DiskError::DiskNotFound)
+                ),
                 "{state}"
             );
             assert!(!ctx.namespace_commits_pending());

crates/scanner/src/scanner.rs

@@ -3575,16 +3575,21 @@ async fn scanner_local_publication_defer_reason<S>(storeapi: &S, observational_o
 where
     S: ScannerStorage,
 {
+    // An authoritative publication is blocked by movement or by a namespace
+    // commit that cannot be attributed to a bucket. An attributed commit that
+    // began after the walk's pending set was read is the next cycle's work,
+    // like a write after publication: the next walk holds its bucket grey if
+    // the commit overlaps that walk, so it does not defer this publication.
     let blocked = if observational_only {
         storeapi.scanner_data_movement_publication_blocked().await
     } else {
-        storeapi.scanner_data_usage_publication_blocked().await
+        storeapi.scanner_attributed_usage_publication_blocked().await
     };
     if !blocked {
         return None;
     }
-    // Pending namespace commits invalidate this publication attempt, but only
-    // storage movement creates durable, rate-limited catch-up debt.
+    // Unattributed namespace commits invalidate this publication attempt, but
+    // only storage movement creates durable, rate-limited catch-up debt.
     if storeapi.scanner_data_movement_publication_blocked().await {
         debug!(
             target: "rustfs::scanner",
@@ -3603,7 +3608,7 @@ where
             component = LOG_COMPONENT_SCANNER,
             subsystem = LOG_SUBSYSTEM_RUNTIME,
             stage = "local_barrier",
-            blocker = "pending_namespace_commit",
+            blocker = "unattributed_namespace_commit",
             "Scanner usage publication deferred"
         );
         Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)

crates/scanner/src/scanner_io.rs

@@ -1127,13 +1127,15 @@ async fn scanner_cycle_activity_status<S>(
     store: &S,
     distributed: bool,
     before: &crate::scanner::ScannerActivitySnapshot,
+    namespace_commits: &ScannerCycleNamespaceCommits,
 ) -> (ScannerCycleActivityStatus, Vec<(String, String, u64)>)
 where
     S: ScannerStorage,
 {
-    // Read the pending-commit barrier before sampling its completion generation.
-    // A tail that drains during this await must invalidate the earlier baseline.
-    let publication_blocked = store.scanner_data_usage_publication_blocked().await;
+    // Namespace commits that overlap the walk and name their bucket keep that
+    // bucket grey (`ScannerCycleNamespaceCommits::held`). Only movement or a
+    // commit that cannot be attributed to a bucket remains a local barrier.
+    let publication_blocked = namespace_commits.unattributed || store.scanner_attributed_usage_publication_blocked().await;
     match crate::scanner::probe_scanner_activity(store, distributed).await {
         Ok(after) => {
             let veto = if publication_blocked {
@@ -1196,6 +1198,64 @@ where
     }
 }
 
+/// Namespace commits that overlap one walk: in flight at the post-walk check
+/// or ended after the floor the cycle recorded before walking.
+#[derive(Debug, Default)]
+pub(super) struct ScannerCycleNamespaceCommits {
+    /// Listed buckets kept grey: reported pending and left out of this
+    /// cycle's certification.
+    held: BTreeSet<String>,
+    /// Some overlapping commit could not be attributed to a bucket.
+    unattributed: bool,
+}
+
+impl ScannerCycleNamespaceCommits {
+    /// Add the held buckets to a `Complete` cycle's pending set.
+    fn extend_pending(&self, pending: &mut ScannerCyclePendingBuckets) {
+        pending.extend(self.held.iter().cloned());
+    }
+
+    /// Keep the held buckets out of the certification snapshot.
+    fn exclude_from_clear(&self, clear: Option<DirtyUsageBuckets>) -> Option<DirtyUsageBuckets> {
+        clear.map(|mut clear| {
+            clear.retain(|bucket, _| !self.held.contains(bucket));
+            clear
+        })
+    }
+}
+
+/// Read, after the walk, every namespace commit that overlaps it and hold its
+/// bucket grey. See "In-flight namespace commits" in
+/// docs/architecture/scanner-per-bucket-convergence.md.
+fn observe_post_walk_namespace_commits<S>(store: &S, floor: u64, listed_buckets: &[BucketInfo]) -> ScannerCycleNamespaceCommits
+where
+    S: ScannerStorage,
+{
+    let cards = store.scanner_namespace_commit_cards(floor);
+    let listed = listed_buckets
+        .iter()
+        .map(|bucket| bucket.name.as_str())
+        .collect::<HashSet<_>>();
+    let held = hold_namespace_commit_buckets(&cards, &listed);
+    if !held.is_empty() || cards.unattributed {
+        debug!(
+            target: "rustfs::scanner::io",
+            event = EVENT_SCANNER_SET_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_IO,
+            state = "cycle_namespace_commits_held",
+            stage = "post_scan_probe",
+            held_buckets = held.len(),
+            unattributed = cards.unattributed,
+            "Scanner kept buckets with namespace commits overlapping the walk grey"
+        );
+    }
+    ScannerCycleNamespaceCommits {
+        held,
+        unattributed: cards.unattributed,
+    }
+}
+
 fn scanner_results_have_pending_maintenance_work(results: &[DataUsageCache]) -> bool {
     results.iter().any(|result| !result.info.pending_heals.is_empty())
 }

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -537,6 +537,54 @@ mod status_class_mutation_tests {
         );
         clear_dirty_usage_buckets_for_tests();
     }
+
+    #[test]
+    #[serial]
+    fn namespace_commit_hold_keeps_marked_buckets_untouched_and_shades_unmarked_ones() {
+        use crate::storage_api::{EcstoreScannerNamespaceCommitCardScope as Scope, EcstoreScannerNamespaceCommitCards};
+
+        clear_dirty_usage_buckets_for_tests();
+        record_dirty_usage_object_from_producer("marked", "2026/a.jpg", SegmentInvalidationProducerIdentity::PutObject);
+        let marked_generation = dirty_usage_buckets_for_tests()["marked"];
+        let cards = EcstoreScannerNamespaceCommitCards {
+            buckets: BTreeMap::from([
+                (
+                    "marked".to_string(),
+                    Scope::Objects(BTreeSet::from(["2026/a.jpg".to_string(), "heal/c.jpg".to_string()])),
+                ),
+                ("healed".to_string(), Scope::Objects(BTreeSet::from(["2026/b.jpg".to_string()]))),
+                ("unlisted".to_string(), Scope::WholeBucket),
+            ]),
+            unattributed: false,
+        };
+
+        let held = hold_namespace_commit_buckets(&cards, &HashSet::from(["marked", "healed"]));
+
+        assert_eq!(held, BTreeSet::from(["healed".to_string(), "marked".to_string()]));
+        let dirty = dirty_usage_buckets_for_tests();
+        assert_eq!(
+            dirty["marked"], marked_generation,
+            "a marked bucket keeps the generation its S3 mark recorded"
+        );
+        assert!(
+            dirty_usage_producer_identities()["marked"].fully_identified,
+            "holding a marked bucket must not erase its producer identity"
+        );
+        assert_eq!(
+            dirty_usage_bucket_scopes()["marked"],
+            DirtyUsageBucketScope::TopLevelEntries(HashSet::from(["2026".to_string(), "heal".to_string()])),
+            "a marked bucket's scope widens to the unmarked commit's entry"
+        );
+        assert!(dirty["healed"] > marked_generation, "an unmarked bucket is shaded");
+        assert!(!dirty_usage_producer_identities()["healed"].fully_identified);
+        assert_eq!(
+            dirty_usage_bucket_scopes()["healed"],
+            DirtyUsageBucketScope::TopLevelEntries(HashSet::from(["2026".to_string()])),
+            "the shade keeps the commit's top-level entry"
+        );
+        assert!(!dirty.contains_key("unlisted"), "an unwalked bucket is left to its own marks");
+        clear_dirty_usage_buckets_for_tests();
+    }
 }
 
 #[cfg(test)]
@@ -1277,6 +1325,78 @@ pub(super) fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo], absent_genera
     }
 }
 
+/// Hold the listed buckets whose namespace commits overlap the walk grey.
+///
+/// A commit's S3-layer dirty mark can precede its last disk effect (an
+/// early-ACK rename tail), so a bucket already in the dirty map may have been
+/// walked mid-commit without its generation moving. Such a bucket keeps the
+/// generation and producer identities its S3 marks recorded; only its scope is
+/// widened to the commit's top-level entries, because a commit without a mark
+/// (a heal rewrite) may touch an entry the marked scope does not cover. The
+/// caller reports it pending and keeps it out of this cycle's certification.
+/// A bucket not in the dirty map has no mark yet, or never gets one; it is
+/// shaded here without a producer identity so the next cycle walks it.
+/// Buckets outside `listed` were not walked and are left to their own marks.
+pub(super) fn hold_namespace_commit_buckets(
+    cards: &crate::storage_api::EcstoreScannerNamespaceCommitCards,
+    listed: &HashSet<&str>,
+) -> BTreeSet<String> {
+    use crate::segment_invalidation::SegmentInvalidationProducerIdentity::Unknown;
+    use crate::storage_api::EcstoreScannerNamespaceCommitCardScope as Scope;
+
+    let mut held = BTreeSet::new();
+    for (bucket, scope) in &cards.buckets {
+        if !listed.contains(bucket.as_str()) {
+            continue;
+        }
+        if !widen_dirty_usage_bucket_scope(bucket, scope) {
+            match scope {
+                Scope::Objects(objects) if !objects.is_empty() => {
+                    for object in objects {
+                        record_dirty_usage_object_from_producer(bucket, object, Unknown);
+                    }
+                }
+                _ => record_dirty_usage_bucket_from_producer(bucket, Unknown),
+            }
+        }
+        held.insert(bucket.clone());
+    }
+    held
+}
+
+/// Widen a dirty bucket's scope to cover a namespace-commit card without moving
+/// its generation or producer identities. Returns `false` when the bucket is
+/// not dirty. The widening is process-local, like the commit card itself.
+fn widen_dirty_usage_bucket_scope(bucket: &str, card: &crate::storage_api::EcstoreScannerNamespaceCommitCardScope) -> bool {
+    use crate::storage_api::EcstoreScannerNamespaceCommitCardScope as Scope;
+
+    // Same lock order as `record_dirty_usage_object_inner`.
+    let dirty_buckets = dirty_usage_buckets();
+    if !dirty_buckets.contains_key(bucket) {
+        return false;
+    }
+    let mut dirty_scopes = dirty_usage_bucket_scopes();
+    let scope = dirty_scopes
+        .entry(bucket.to_string())
+        .or_insert(DirtyUsageBucketScope::WholeBucket);
+    let DirtyUsageBucketScope::TopLevelEntries(entries) = scope else {
+        return true;
+    };
+    let widened = match card {
+        Scope::Objects(objects) if !objects.is_empty() => objects.iter().all(|object| {
+            dirty_usage_top_level_entry(object).is_some_and(|entry| {
+                entries.insert(entry);
+                entries.len() <= MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET
+            })
+        }),
+        _ => false,
+    };
+    if !widened {
+        *scope = DirtyUsageBucketScope::WholeBucket;
+    }
+    true
+}
+
 pub(crate) fn dirty_usage_buckets_pending() -> bool {
     !dirty_usage_buckets().is_empty()
 }

crates/scanner/src/scanner_io/io_cycle.rs

@@ -366,6 +366,8 @@ where
             ));
         }
     };
+    // Commits that end at or below this floor end before the walk reads.
+    let namespace_commit_floor = store.scanner_namespace_commit_card_floor();
     let dirty_generation_before_bucket_list = dirty_usage_generation();
     let bucket_listing = store.list_bucket_for_scanner(&BucketOptions::default()).await?;
     let mut bucket_plan_complete = bucket_listing.topology_complete;
@@ -440,8 +442,9 @@ where
         if !bucket_plan_complete {
             return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch));
         }
+        let namespace_commits = observe_post_walk_namespace_commits(store, namespace_commit_floor, &all_buckets);
         let (activity_status, remote_publication_lease_targets) =
-            scanner_cycle_activity_status(store, distributed, &activity_before).await;
+            scanner_cycle_activity_status(store, distributed, &activity_before, &namespace_commits).await;
         let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot);
         let status = classify_nsscanner_cycle(
             true,
@@ -452,7 +455,9 @@ where
             activity_status,
         );
         let pending_buckets = if status == ScannerCycleStatus::Complete {
-            pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets)
+            let mut pending = pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets);
+            namespace_commits.extend_pending(&mut pending);
+            pending
         } else {
             ScannerCyclePendingBuckets::new()
         };
@@ -482,7 +487,8 @@ where
         if status == ScannerCycleStatus::Complete {
             complete_tier_registry_cycle(want_cycle, leader_epoch);
         }
-        let dirty_usage_clear = (status == ScannerCycleStatus::Complete).then(|| dirty_usage_snapshot.buckets.as_ref().clone());
+        let dirty_usage_clear = namespace_commits
+            .exclude_from_clear((status == ScannerCycleStatus::Complete).then(|| dirty_usage_snapshot.buckets.as_ref().clone()));
         let remote_dirty_usage_acknowledgements = if status == ScannerCycleStatus::Complete {
             crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before)
         } else {
@@ -765,8 +771,9 @@ where
             !namespace_not_found_buckets.is_empty(),
         );
     let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot);
+    let namespace_commits = observe_post_walk_namespace_commits(store, namespace_commit_floor, &all_buckets);
     let (activity_status, remote_publication_lease_targets) =
-        scanner_cycle_activity_status(store, distributed, &activity_before).await;
+        scanner_cycle_activity_status(store, distributed, &activity_before, &namespace_commits).await;
     let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
     let completed_usage = completed_usage_candidate(
         &results,
@@ -813,7 +820,9 @@ where
     // guardrail verdict so a bucket written during the walk is reported in
     // the same publication that certifies the others.
     let pending_buckets = if cycle_status == ScannerCycleStatus::Complete {
-        pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets)
+        let mut pending = pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets);
+        namespace_commits.extend_pending(&mut pending);
+        pending
     } else {
         ScannerCyclePendingBuckets::new()
     };
@@ -833,14 +842,14 @@ where
     } else {
         false
     };
-    let dirty_usage_clear = should_clear_dirty_usage_snapshot(
+    let dirty_usage_clear = namespace_commits.exclude_from_clear(should_clear_dirty_usage_snapshot(
         result.is_ok(),
         structurally_complete_snapshot,
         budget_elapsed,
         activity_status == ScannerCycleActivityStatus::Unchanged,
         &dirty_usage_snapshot.buckets,
         &failed_buckets,
-    );
+    ));
     result?;
     if cycle_status == ScannerCycleStatus::Complete {
         complete_tier_registry_cycle(want_cycle, leader_epoch);

crates/scanner/src/scanner_io/tests.rs

@@ -23,7 +23,7 @@ use crate::scanner_folder::ScannerItem;
 use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
 use crate::storage_api::owner::{
     EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
-    WriteCommitGuard, WriteCompletion, ecstore_hold_namespace_commit,
+    WriteCommitGuard, WriteCompletion, ecstore_hold_bucket_namespace_commit, ecstore_hold_namespace_commit,
 };
 use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
 use crate::{

crates/scanner/src/scanner_io/tests/per_bucket_certification.rs

@@ -4,7 +4,9 @@
 //! Per-bucket certification on the single-disk fixture: a bucket written
 //! during the walk is reported pending while the other buckets are certified,
 //! the publication stays authoritative, and the next cycle is scoped to the
-//! pending bucket. A guardrail veto still supersedes the whole cycle.
+//! pending bucket. An in-flight namespace commit that names its bucket only
+//! shades that bucket; a guardrail veto, including a commit that cannot be
+//! attributed, still supersedes the whole cycle.
 
 use super::*;
 
@@ -196,7 +198,7 @@ async fn write_during_walk_leaves_only_that_bucket_pending_and_scopes_the_next_c
 
 #[tokio::test]
 #[serial]
-async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() {
+async fn held_put_tail_leaves_only_its_bucket_pending_instead_of_superseding() {
     let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
     clear_dirty_usage_buckets_for_tests();
     for bucket in BUCKETS {
@@ -210,8 +212,179 @@ async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() {
     wait_for_namespace_commit_tails(store.as_ref()).await;
     let vetoes_before = global_metrics().report().await.usage_freshness.guardrail_veto_total;
 
-    // A PUT tail held across the walk flips the local publication barrier,
-    // which is a guardrail veto regardless of the dirty map.
+    // A continuously written single node always has some rename tail in
+    // flight. Hold one across the whole cycle, publication included: it
+    // names its bucket, so it shades that bucket instead of vetoing.
+    let pending_commit = ecstore_hold_bucket_namespace_commit(store.as_ref(), "hot-bucket", "tail/object");
+    assert!(store.scanner_data_usage_publication_blocked().await);
+    let (result, first) = run_cycle(
+        &store,
+        1,
+        None,
+        true,
+        CycleHooks {
+            resolved_scope_observer: None,
+            post_walk_gate: None,
+        },
+    )
+    .await;
+
+    assert_eq!(
+        result.status,
+        ScannerCycleStatus::Complete,
+        "an attributed in-flight commit no longer supersedes the cycle"
+    );
+    assert_eq!(result.pending_dirty_usage_buckets(), 1);
+    assert!(!first.usage_snapshot_observational, "pending buckets publish on the authoritative route");
+    assert_eq!(first.usage_snapshot_converged, Some(false));
+    assert_eq!(first.usage_snapshot_pending_buckets, vec!["hot-bucket".to_string()]);
+    assert!(first.is_complete_bucket_usage_snapshot());
+    for bucket in BUCKETS {
+        assert_eq!(first.buckets_usage[bucket].objects_count, 1, "{bucket} carries its walked value");
+    }
+    assert!(result.has_dirty_usage_to_acknowledge());
+    assert!(result.clear_verified_usage().is_empty());
+    assert_eq!(
+        dirty_usage_buckets_for_tests().keys().cloned().collect::<Vec<_>>(),
+        vec!["hot-bucket".to_string()],
+        "certification clears the quiet buckets and keeps the bucket with the in-flight commit"
+    );
+    assert_eq!(
+        global_metrics().report().await.usage_freshness.guardrail_veto_total,
+        vetoes_before,
+        "an attributed commit is not a guardrail veto"
+    );
+
+    // The tail drains between cycles, before the next walk records its
+    // floor, so that cycle walks only the hot bucket and converges.
+    drop(pending_commit);
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+    let (observer, resolved_scope) = tokio::sync::oneshot::channel();
+    let (result, second) = run_cycle(
+        &store,
+        2,
+        Some(baseline_bytes(&first)),
+        false,
+        CycleHooks {
+            resolved_scope_observer: Some(observer),
+            post_walk_gate: None,
+        },
+    )
+    .await;
+    let scope = resolved_scope.await.expect("the cycle should report its resolved scope");
+    assert_eq!(
+        scope.selected_buckets_for_tests(),
+        Some(&HashSet::from(["hot-bucket".to_string()])),
+        "the next cycle walks only the bucket the tail shaded"
+    );
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    assert_eq!(result.pending_dirty_usage_buckets(), 0);
+    assert_eq!(second.usage_snapshot_converged, Some(true));
+    assert!(second.usage_snapshot_pending_buckets.is_empty());
+    assert_eq!(second.buckets_usage["hot-bucket"].objects_count, 1);
+    assert!(result.clear_verified_usage().is_empty());
+    assert!(dirty_usage_buckets_for_tests().is_empty());
+
+    // A commit with no S3 dirty mark (a heal rewrite) on a bucket that is not
+    // dirty: holding it pending is not enough, so the collector shades it and
+    // the following cycle walks it.
+    let heal_commit = ecstore_hold_bucket_namespace_commit(store.as_ref(), "idle-bucket", "healed/object");
+    let (result, third) = run_cycle(
+        &store,
+        3,
+        Some(baseline_bytes(&second)),
+        false,
+        CycleHooks {
+            resolved_scope_observer: None,
+            post_walk_gate: None,
+        },
+    )
+    .await;
+    drop(heal_commit);
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    assert_eq!(third.usage_snapshot_pending_buckets, vec!["idle-bucket".to_string()]);
+    assert!(result.clear_verified_usage().is_empty());
+    assert_eq!(
+        dirty_usage_buckets_for_tests().keys().cloned().collect::<Vec<_>>(),
+        vec!["idle-bucket".to_string()],
+        "an unmarked commit's bucket is shaded into the dirty map"
+    );
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn put_tail_that_ends_during_the_walk_leaves_its_bucket_pending() {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
+    clear_dirty_usage_buckets_for_tests();
+    for bucket in BUCKETS {
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        put_fixture_object(&store, bucket, "initial").await;
+        record_dirty_usage_bucket(bucket);
+    }
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+
+    // The S3 layer marks a PUT dirty when the request returns, which for an
+    // early-ACK PUT is before its rename tail ends. A tail that ends during
+    // the walk therefore never moves the generation certification compares;
+    // its ended card is what keeps the bucket grey.
+    let pending_commit = ecstore_hold_bucket_namespace_commit(store.as_ref(), "hot-bucket", "tail/object");
+    let (observer, resolved_scope) = tokio::sync::oneshot::channel();
+    let (release, gate) = tokio::sync::oneshot::channel();
+    let ender = tokio::spawn(async move {
+        resolved_scope.await.expect("the cycle should report its resolved scope");
+        drop(pending_commit);
+        release.send(()).expect("the cycle should still be waiting on the gate");
+    });
+    let (result, snapshot) = run_cycle(
+        &store,
+        1,
+        None,
+        true,
+        CycleHooks {
+            resolved_scope_observer: Some(observer),
+            post_walk_gate: Some(gate),
+        },
+    )
+    .await;
+    ender.await.expect("the tail owner task should finish");
+
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    assert_eq!(snapshot.usage_snapshot_pending_buckets, vec!["hot-bucket".to_string()]);
+    assert!(result.clear_verified_usage().is_empty());
+    assert_eq!(
+        dirty_usage_buckets_for_tests().keys().cloned().collect::<Vec<_>>(),
+        vec!["hot-bucket".to_string()],
+        "the tail's end shades its bucket even though the S3 mark predates the walk"
+    );
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn unattributed_namespace_commit_still_supersedes_the_cycle_onto_the_observed_route() {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
+    clear_dirty_usage_buckets_for_tests();
+    for bucket in BUCKETS {
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        put_fixture_object(&store, bucket, "initial").await;
+        record_dirty_usage_bucket(bucket);
+    }
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+    let vetoes_before = global_metrics().report().await.usage_freshness.guardrail_veto_total;
+
+    // Every production commit names its bucket, so a held PUT tail now only
+    // shades that bucket (see the held-tail test above). A commit that cannot
+    // be attributed (an owner without a bucket name, an exhausted commit
+    // counter, or an end that did not fit the card table) still flips the
+    // local publication barrier, which is a guardrail veto regardless of the
+    // dirty map.
     let pending_commit = ecstore_hold_namespace_commit(store.as_ref());
     assert!(store.scanner_data_usage_publication_blocked().await);
     let (result, snapshot) = run_cycle(

crates/scanner/src/storage_api.rs

@@ -134,6 +134,8 @@ pub(crate) use rustfs_ecstore::api::storage::SCANNER_PUBLICATION_LEASE_TTL_MS as
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
 pub(crate) use rustfs_ecstore::api::storage::{
     ECStore as EcstoreStore, ScannerDataMovementPauseStatus as EcstoreScannerDataMovementPauseStatus,
+    ScannerNamespaceCommitCardScope as EcstoreScannerNamespaceCommitCardScope,
+    ScannerNamespaceCommitCards as EcstoreScannerNamespaceCommitCards,
 };
 pub(crate) use rustfs_lifecycle::{
     Evaluator as EcstoreEvaluator, Event as EcstoreEvent, Lifecycle as EcstoreLifecycle, ObjectOpts as EcstoreObjectOpts,
@@ -169,7 +171,8 @@ pub(crate) mod owner {
     pub(crate) use rustfs_ecstore::api::set_disk::get_lock_acquire_timeout as ecstore_get_lock_acquire_timeout;
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::set_disk::test_util::{
-        PutObjectCommitBarrier, PutObjectCommitPause, hold_namespace_commit as ecstore_hold_namespace_commit,
+        PutObjectCommitBarrier, PutObjectCommitPause, hold_bucket_namespace_commit as ecstore_hold_bucket_namespace_commit,
+        hold_namespace_commit as ecstore_hold_namespace_commit,
     };
 
     pub(crate) use super::storage_contracts::{
@@ -390,7 +393,9 @@ pub(crate) trait ScannerStorage:
     fn scanner_topology_digest(&self) -> [u8; 32];
     fn scanner_namespace_mutation_generation(&self) -> u64;
     async fn scanner_data_movement_activity(&self) -> (bool, bool, u64);
-    async fn scanner_data_usage_publication_blocked(&self) -> bool;
+    async fn scanner_attributed_usage_publication_blocked(&self) -> bool;
+    fn scanner_namespace_commit_card_floor(&self) -> u64;
+    fn scanner_namespace_commit_cards(&self, floor: u64) -> EcstoreScannerNamespaceCommitCards;
     async fn scanner_data_movement_publication_blocked(&self) -> bool;
     async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus;
     fn scanner_data_movement_generation(&self) -> u64;
@@ -427,8 +432,16 @@ impl ScannerStorage for EcstoreStore {
         EcstoreStore::scanner_data_movement_activity(self).await
     }
 
-    async fn scanner_data_usage_publication_blocked(&self) -> bool {
-        EcstoreStore::scanner_data_usage_publication_blocked(self).await
+    async fn scanner_attributed_usage_publication_blocked(&self) -> bool {
+        EcstoreStore::scanner_attributed_usage_publication_blocked(self).await
+    }
+
+    fn scanner_namespace_commit_card_floor(&self) -> u64 {
+        EcstoreStore::scanner_namespace_commit_card_floor(self)
+    }
+
+    fn scanner_namespace_commit_cards(&self, floor: u64) -> EcstoreScannerNamespaceCommitCards {
+        EcstoreStore::scanner_namespace_commit_cards(self, floor)
     }
 
     async fn scanner_data_movement_publication_blocked(&self) -> bool {
@@ -600,10 +613,18 @@ mod tests {
             (false, false, 0)
         }
 
-        async fn scanner_data_usage_publication_blocked(&self) -> bool {
+        async fn scanner_attributed_usage_publication_blocked(&self) -> bool {
             false
         }
 
+        fn scanner_namespace_commit_card_floor(&self) -> u64 {
+            0
+        }
+
+        fn scanner_namespace_commit_cards(&self, _floor: u64) -> EcstoreScannerNamespaceCommitCards {
+            EcstoreScannerNamespaceCommitCards::default()
+        }
+
         async fn scanner_data_movement_publication_blocked(&self) -> bool {
             false
         }

docs/architecture/scanner-per-bucket-convergence.md

@@ -1,7 +1,7 @@
 # Scanner Per-Bucket Convergence Design
 
 **Use this when:** changing what makes a scanner cycle `Complete`, `Superseded`, or converged; adding or re-classifying a dirty-usage producer; deciding whether a background writeback may invalidate a scanner cycle; changing scanner worker concurrency or throttling; or activating scoped dirty acknowledgements.
-**Source of truth:** `classify_nsscanner_cycle`, `scanner_cycle_activity_status`, and `scoped_scan_scope_from_dirty_buckets` in `crates/scanner/src/scanner_io.rs`; `dirty_usage_snapshot_status` and `acknowledge_scoped_dirty_usage` in `crates/scanner/src/scanner_io/dirty_usage.rs`; `scanner_concurrency_limit` in `crates/scanner/src/scanner_io/guards.rs`; the per-disk worker loop in `crates/scanner/src/scanner_io/io_cache.rs`; `DynamicSleeper` in `crates/scanner/src/sleeper.rs`; `ScannerNodeActivity` in `crates/scanner/src/scanner/activity.rs`; `persist_replication_state_if_current` in `crates/ecstore/src/bucket/replication/replication_resyncer.rs`; the publication fences in [scanner-usage-publication.md](scanner-usage-publication.md).
+**Source of truth:** `classify_nsscanner_cycle`, `scanner_cycle_activity_status`, and `scoped_scan_scope_from_dirty_buckets` in `crates/scanner/src/scanner_io.rs`; `dirty_usage_snapshot_status` and `acknowledge_scoped_dirty_usage` in `crates/scanner/src/scanner_io/dirty_usage.rs`; `scanner_concurrency_limit` in `crates/scanner/src/scanner_io/guards.rs`; the per-disk worker loop in `crates/scanner/src/scanner_io/io_cache.rs`; `DynamicSleeper` in `crates/scanner/src/sleeper.rs`; `ScannerNodeActivity` in `crates/scanner/src/scanner/activity.rs`; `persist_replication_state_if_current` in `crates/ecstore/src/bucket/replication/replication_resyncer.rs`; `NamespaceCommitCardTable` in `crates/ecstore/src/runtime/namespace_commit_cards.rs` and `observe_post_walk_namespace_commits` in `crates/scanner/src/scanner_io.rs`; the publication fences in [scanner-usage-publication.md](scanner-usage-publication.md).
 
 ## Decision
 
@@ -19,9 +19,11 @@ collection, because the problems and the known-good answers are the same.
    turn the cycle into `Superseded`.
 2. **The cluster activity probe stays, as a guardrail only.** It vetoes a
    publication when the heap it describes is not the heap that was walked:
-   topology, peer identity, protocol version, data movement, or a pending PUT
-   tail changed. Namespace and dirty-usage generation movement alone never
-   vetoes; it only shades buckets.
+   topology, peer identity, protocol version, or data movement changed, or a
+   namespace commit is in flight that cannot be attributed to a bucket.
+   Namespace and dirty-usage generation movement alone never vetoes, and
+   neither does an in-flight commit that names its bucket; both only shade
+   buckets.
 3. **Replication status writebacks are status-class barrier hits.** They
    shade a bucket for a scoped rescan but are not namespace activity. They
    never uncertify a walk and never block cycle completion. The scanner's own
@@ -206,7 +208,7 @@ walk. Field roles:
 | Field | Role |
 |---|---|
 | `instance_id`, `protocol_version`, `topology_digest` | veto: a peer restarted, changed version, or the topology changed |
-| `data_movement_active`, `movement_generation`, `publication_blocked` | veto: rebalance, decommission, or a pending PUT tail can make the walk describe a different generation |
+| `data_movement_active`, `movement_generation`, `publication_blocked` | veto: rebalance or decommission can make the walk describe a different generation. On the wire `publication_blocked` carries only the movement barrier; in-flight namespace commits are not part of any field and are handled below |
 | `maintenance_generation` | veto: no cycle today records whether it published maintenance results, and the field is already part of the structural digest; it becomes a shade signal only once that distinction exists |
 | `namespace_generation`, `dirty_usage_generation`, `dirty_usage_pending` | shade signal: consumed per bucket; never a veto on their own |
 
@@ -239,6 +241,75 @@ The guardrail is bounded in three ways:
   field and peer that caused the last deferral so the operator can tell a
   rebalance from a broken probe.
 
+### In-flight namespace commits
+
+A namespace commit (PUT and multipart rename fanouts including the post-ACK
+tail, delete commits, delete recovery, heal rewrites, and the peer-RPC disk
+writes of a remote coordinator) holds a
+`NamespaceCommitGuard` from before its first disk effect until its last disk
+task ends. The S3 layer marks the bucket dirty when the request returns,
+which for an early-ACK PUT is before the tail ends, and heal rewrites mark
+nothing. A walk that overlaps such a commit therefore may read partial state
+for a bucket whose dirty generation never moves afterwards, so an in-flight
+commit cannot simply be ignored.
+
+Before this rule the local barrier was the instance-wide pending-commit count:
+any commit in flight at the post-walk check vetoed the cycle as
+`publication_blocked`, and any commit in flight at publication deferred it. A
+single node under continuous writes always has a commit in flight, so it
+never published (the #8305 class of guardrail false positive), while a tail
+that ended between the walk and the check was not caught at all.
+
+The guard is now a card on the bucket it commits to, kept by the instance
+context next to the count (`NamespaceCommitCardTable` in
+`crates/ecstore/src/runtime/namespace_commit_cards.rs`):
+
+- **Begin.** Every production commit is opened with its bucket and, where it
+  has one, its object (`begin_bucket_namespace_commit`). It is registered as
+  in flight before any disk effect. Meta-bucket commits are counted but not
+  carded; the scanner does not walk meta buckets for usage.
+- **End.** When the last owner of the guard drops, after the last disk task,
+  the bucket is stamped with a monotonic end sequence. Objects are kept per
+  bucket up to the dirty map's top-level-entry bound, and across the table up
+  to a fixed total, and then collapse to the whole bucket.
+- **Collector.** Before listing buckets, the cycle records the end sequence
+  as its floor. After the walk, before the guardrail verdict and the pending
+  read, it asks for every bucket with a commit in flight or ended above the
+  floor (`observe_post_walk_namespace_commits`). Each such listed bucket is
+  held grey: it is added to the pending set and left out of this cycle's
+  certification. A held bucket that is not in the dirty map is also shaded
+  into it without a producer identity, since it has no S3 mark yet or never
+  gets one (a heal rewrite), so the next cycle walks it. A held bucket that
+  is already dirty keeps the generation and producer identities its S3 marks
+  recorded, so segment-reuse evidence is not weakened for buckets under
+  ordinary writes; only its scope widens to the commit's top-level entries,
+  because an unmarked commit may touch an entry the marked scope does not
+  cover.
+
+Correctness, against the floor `F` (read before the walk), the start snapshot
+`S`, and the post-walk read `P`:
+
+- a commit that ended at or below `F` ended before the walk read anything,
+  so the walk saw all of its effects;
+- a commit that ended above `F` and before `P`, or is in flight at `P`, is
+  held: the bucket is pending and certification does not clear it. If it was
+  dirty it stays dirty at its marked generation; if it was not, it is
+  shaded. Either way the next cycle walks it, and that cycle certifies it
+  only if no commit overlaps its own walk;
+- a commit that begins after `P` is the next cycle's work, like a write after
+  publication: its end is above the next floor only if it overlaps the next
+  walk, and its S3 mark, if any, moves the generation certification
+  compares.
+
+Certification is compare-and-remove against `S`, so no commit that overlaps a
+walk can leave its bucket black. The local barrier keeps only what cannot be
+attributed: a guard opened without a bucket name, counter exhaustion, or an
+end that did not fit the card table. Every production commit, including the
+peer-RPC admission of a remote coordinator's disk write, names its bucket. Those still veto the post-walk check as `publication_blocked`
+(an unscoped guard or exhaustion also defers publication), which is the
+previous behaviour. Distributed publication leases on peers still reject
+pending commits; attributing them is phase 5 work.
+
 ### Generations of buckets
 
 Most buckets are cold and a few are hot. The scanner keeps two populations
@@ -318,6 +389,7 @@ not tuning advice.
 | `Status` hits are not namespace activity | A `Status` hit costs one atomic OR on the writeback path and no scanner work. It never triggers a dirty-usage wake. A replication resync therefore adds nothing to scanner load beyond the full walk the cycle already performs. |
 | A `Complete` cycle with pending buckets is a baseline | When the pending set is the whole namespace (a mass write, a disk loss), the next cycle is declared a full sweep explicitly and uses the existing superseded retry backoff schedule. A full sweep is never run under the scoped cycle's wake cadence. |
 | Bounded remark | At most one remark pass per cycle, inside the remaining budget. |
+| An attributed in-flight commit holds its bucket grey instead of vetoing | A cycle holds at most its listed buckets, once each, and shades only those not already dirty, each scoped to at most the dirty map's top-level-entry bound. The card table holds at most a fixed number of ended buckets and object names, whether or not the scanner runs, and prunes everything at or below the latest floor on every post-walk read; an end that does not fit, an unscoped guard, or counter exhaustion falls back to the whole-cycle veto with the capped superseded backoff. A permanently written bucket stays grey and is rescanned under the ordinary dirty-usage wake; holding adds no wake of its own, and a shade adds at most one per unmarked bucket per cycle. |
 | Old-generation sweep every `M` cycles | The sweep is checkpointed and resumes; it does not restart on a `Namespace` hit. A hit only demotes the bucket to young for the following cycle. |
 | Pacer raises walkers and lowers sleep | CPU and I/O ceilings are absolute. Foreground admission backoff cannot be undercut. Walker count never exceeds the ceiling-derived maximum even when the target is missed. |
 | Faster convergence discovers heal and replication work sooner | Heal admission keeps its queue bounds and `queue_full` and `dropped` accounting in `crates/scanner/src/scanner_folder/ledger.rs`; replication MRF keeps its pending cap. Convergence speed does not change retry intervals. Discovered work is swept lazily under those bounds, in the sense of a lazy sweeper: discovery and execution are decoupled. |

docs/architecture/scanner-usage-publication.md

@@ -27,8 +27,12 @@ Ordinary PUT rename fanouts also track instance-scoped in-flight work. A quorum
 ACK does not release it: the actual disk tasks retain ownership until their
 rename work ends, including when the request caller is cancelled. Scan admission
 remains movement-only so sustained PUTs do not stop namespace walks and
-scanner-driven lifecycle discovery. The post-walk local publication check and
-remote publication leases reject pending fanouts. Begin/end namespace generations
+scanner-driven lifecycle discovery. Locally, a fanout names its bucket: the
+post-walk check holds that bucket grey (pending, not certified) instead of
+rejecting the cycle, and only a commit that cannot be attributed to a bucket still blocks
+local publication (see "In-flight namespace commits" in
+[scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md)).
+Remote publication leases still reject pending fanouts. Begin/end namespace generations
 invalidate cached plans across the fanout; after acquiring remote leases, the
 coordinator rechecks the guardrail fields of the activity snapshot (the
 structural digest and the movement fields, plus the peers' dirty-usage state
@@ -40,8 +44,9 @@ are the next cycle's work, as described in
 [scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md).
 
 This adds no namespace or movement lock. An already-verified older snapshot may
-still precede a newly started write. Sustained or stalled PUT tails can delay
-authoritative usage publication, which resumes through the existing retry
+still precede a newly started write. On a distributed deployment sustained or
+stalled PUT tails on a peer can delay authoritative usage publication, which
+resumes through the existing retry
 schedule rather than a new immediate-wakeup protocol. Intermediate per-set and
 prefix cache readers retain their existing approximate-cache semantics. A
 prolonged pending tail with no generation changes can also delay cycle advancement