feat(capture): apply native entries atomically with durable replay

Cargo.lock

@@ -275,6 +275,15 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "anyerror"
+version = "0.1.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "71add24cc141a1e8326f249b74c41cfd217aeb2a67c9c6cf9134d175469afd49"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.104"
@@ -1706,6 +1715,30 @@ dependencies = [
  "syn 3.0.6",
 ]
 
+[[package]]
+name = "borsh"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "553c5d846a6ba5150c65e3b1b8ec073bcf1abc20f9b7220de384a4443ea4e20a"
+dependencies = [
+ "borsh-derive",
+ "bytes",
+ "cfg_aliases",
+]
+
+[[package]]
+name = "borsh-derive"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12cdfe656708a01f89b451a7d36466e6fe6c414de0aa18fc54f864f6f9ca9f56"
+dependencies = [
+ "once_cell",
+ "proc-macro-crate",
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
 [[package]]
 name = "brotli"
 version = "8.0.4"
@@ -1772,6 +1805,18 @@ version = "3.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
 
+[[package]]
+name = "byte-unit"
+version = "5.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c719d56f7e96194cfc53460976d3ba51c85719747c9c62ed99981847b551152b"
+dependencies = [
+ "rust_decimal",
+ "schemars 1.2.2",
+ "serde",
+ "utf8-width",
+]
+
 [[package]]
 name = "bytemuck"
 version = "1.25.2"
@@ -3724,13 +3769,34 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "derive_more"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
+dependencies = [
+ "derive_more-impl 1.0.0",
+]
+
 [[package]]
 name = "derive_more"
 version = "2.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134"
 dependencies = [
- "derive_more-impl",
+ "derive_more-impl 2.1.1",
+]
+
+[[package]]
+name = "derive_more-impl"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+ "unicode-xid",
 ]
 
 [[package]]
@@ -6234,7 +6300,7 @@ dependencies = [
  "bytes",
  "chrono",
  "dashmap",
- "derive_more",
+ "derive_more 2.1.1",
  "futures-util",
  "getrandom 0.4.3",
  "lazy_static",
@@ -6388,6 +6454,12 @@ dependencies = [
  "quote",
 ]
 
+[[package]]
+name = "maplit"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
+
 [[package]]
 name = "matchers"
 version = "0.2.0"
@@ -7329,6 +7401,42 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "openraft"
+version = "0.9.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd"
+dependencies = [
+ "anyerror",
+ "byte-unit",
+ "chrono",
+ "clap",
+ "derive_more 1.0.0",
+ "futures",
+ "maplit",
+ "openraft-macros",
+ "rand 0.8.8",
+ "serde",
+ "thiserror 1.0.69",
+ "tokio",
+ "tracing",
+ "tracing-futures",
+ "validit",
+]
+
+[[package]]
+name = "openraft-macros"
+version = "0.9.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "55417d79f46a787dd88c043d831d8b7a0713f7c69d1925f9de51cc42c437250f"
+dependencies = [
+ "chrono",
+ "proc-macro2",
+ "quote",
+ "semver",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "openssl-probe"
 version = "0.2.1"
@@ -8958,6 +9066,15 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "redb"
+version = "4.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb338a6c67830a61bed824b78c2bb034ab1ff401a20616bd7957524f4fdc2f22"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "redis"
 version = "1.7.1"
@@ -9464,6 +9581,23 @@ dependencies = [
  "walkdir",
 ]
 
+[[package]]
+name = "rust_decimal"
+version = "1.43.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7653272e75dcac41dc199fbea6f5797633994fafd339943c06c9af16bf29cd3a"
+dependencies = [
+ "arrayvec",
+ "borsh",
+ "bytes",
+ "num-traits",
+ "rand 0.8.8",
+ "rand 0.9.5",
+ "serde",
+ "serde_json",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "rustc-demangle"
 version = "0.1.28"
@@ -9666,6 +9800,19 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "rustfs-capture-authority"
+version = "1.0.1"
+dependencies = [
+ "openraft",
+ "redb",
+ "serde",
+ "serde_json",
+ "tempfile",
+ "thiserror 2.0.21",
+ "url",
+]
+
 [[package]]
 name = "rustfs-checksums"
 version = "1.0.1"
@@ -10473,6 +10620,7 @@ dependencies = [
  "tokio",
  "tracing",
  "tracing-subscriber",
+ "uuid",
 ]
 
 [[package]]
@@ -13046,6 +13194,16 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "tracing-futures"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
+dependencies = [
+ "pin-project",
+ "tracing",
+]
+
 [[package]]
 name = "tracing-log"
 version = "0.2.0"
@@ -13196,7 +13354,7 @@ checksum = "79acf4b257aa56dd7f00466a81693d6c79c362c55092b3137909d6e76249ad80"
 dependencies = [
  "async-trait",
  "chrono",
- "derive_more",
+ "derive_more 2.1.1",
  "libc",
  "md-5 0.10.6",
  "thiserror 2.0.21",
@@ -13296,6 +13454,12 @@ version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
 
+[[package]]
+name = "utf8-width"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "159a7cadce548703edd50d24069bc294c5415ecab0a480e0cd1ca06d112dc94a"
+
 [[package]]
 name = "utf8_iter"
 version = "1.0.4"
@@ -13322,6 +13486,15 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "validit"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4417835826b78aa6265724624d424df2b1e6e71a33329227d8b04a88950780dc"
+dependencies = [
+ "anyerror",
+]
+
 [[package]]
 name = "valuable"
 version = "0.1.1"

Cargo.toml

@@ -18,6 +18,7 @@ members = [
     "crates/audit", # Audit target management system with multi-target fan-out
     "crates/checksums", # client checksums
     "crates/common", # Shared utilities and data structures
+    "crates/capture-authority", # Durable capture authority storage
     "crates/data-usage", # Shared data usage models and algorithms
     "crates/config", # Configuration management
     "crates/connect-symbol-catalog", # Release-only CPU profile symbol catalog generator
@@ -314,6 +315,8 @@ ratelimit = "2.0.0"
 rayon = "1.12.0"
 rustfs-erasure-codec = { version = "9.0.0" }
 reed-solomon-simd = "3.1.0"
+openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }
+redb = "=4.3.0"
 regex = { version = "1.13.1" }
 rumqttc = { package = "rumqttc-next", version = "0.34.0" }
 redis = { version = "1.7.1" }

crates/capture-authority/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "rustfs-capture-authority"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+repository.workspace = true
+description = "Durable local storage for capture authority"
+
+[dependencies]
+openraft = { workspace = true }
+redb = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+thiserror = { workspace = true }
+url = { workspace = true }
+
+[dev-dependencies]
+serde_json = { workspace = true }
+tempfile = { workspace = true }
+
+[lints]
+workspace = true

crates/capture-authority/src/application.rs

@@ -0,0 +1,826 @@
+use crate::{
+    LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
+    next_index, number,
+};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
+use std::collections::BTreeMap;
+
+pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
+pub(crate) const VERSIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("versions");
+pub(crate) const EVENTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("events");
+pub(crate) const OPERATIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("operations");
+pub(crate) const APPLY_RESULTS: TableDefinition<u64, &[u8]> = TableDefinition::new("apply_results");
+
+fn version_key(object: &ObjectIdentity, version: [u8; 16]) -> Result<Vec<u8>, StoreError> {
+    let mut b = object_bytes(object)?;
+    b.extend(version);
+    Ok(b)
+}
+fn decode_version(b: &[u8]) -> Result<CreatedVersion, StoreError> {
+    let mut d = Decoder::new(b);
+    let v = CreatedVersion {
+        prepared: d.prepared()?,
+        decision_log_id: d.id()?,
+        binding_revision: d.u64()?,
+    };
+    d.finish()?;
+    Ok(v)
+}
+fn decode_event(b: &[u8]) -> Result<CreatedEvent, StoreError> {
+    let mut d = Decoder::new(b);
+    let e = CreatedEvent {
+        object: d.object()?,
+        prepared: d.prepared()?,
+        decision_log_id: d.id()?,
+        binding: d.binding()?,
+    };
+    d.finish()?;
+    Ok(e)
+}
+fn decode_operation(b: &[u8]) -> Result<CreatedOperation, StoreError> {
+    let mut d = Decoder::new(b);
+    let len = usize::try_from(d.u64()?).map_err(|_| StoreError::CorruptRecord)?;
+    let command = b
+        .get(8..8usize.checked_add(len).ok_or(StoreError::CorruptRecord)?)
+        .ok_or(StoreError::CorruptRecord)?;
+    DecideCreated::decode(command)?;
+    let mut d = Decoder::new(&b[8 + len..]);
+    let result = d.result()?;
+    d.finish()?;
+    Ok(CreatedOperation {
+        canonical_command: command.to_vec(),
+        result,
+    })
+}
+fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
+    let mut d = Decoder::new(b);
+    let r = CreatedApplyResult {
+        log_id: d.id()?,
+        result: d.result()?,
+    };
+    d.finish()?;
+    Ok(r)
+}
+fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
+    let b = meta.get("capture_binding")?.ok_or(StoreError::BindingNotPinned)?;
+    let mut d = Decoder::new(b.value());
+    let binding = d.binding()?;
+    d.finish()?;
+    Ok(binding)
+}
+fn read_applied(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<Option<LogId>, StoreError> {
+    let b = meta.get("last_applied")?.ok_or(StoreError::CorruptRecord)?;
+    let mut d = Decoder::new(b.value());
+    let id = match d.take::<1>()?[0] {
+        0 => None,
+        1 => Some(d.id()?),
+        _ => return Err(StoreError::CorruptRecord),
+    };
+    d.finish()?;
+    Ok(id)
+}
+fn required<T>(value: Option<T>) -> Result<T, StoreError> {
+    value.ok_or(StoreError::CorruptRecord)
+}
+fn encode_native_entry(entry: &crate::CaptureRaftEntry) -> Result<Vec<u8>, StoreError> {
+    let id: LogId = entry.log_id.into();
+    let (kind, body) = match &entry.payload {
+        openraft::EntryPayload::Blank => (0, vec![]),
+        openraft::EntryPayload::Normal(command) => (1, command.created().encode()?),
+        openraft::EntryPayload::Membership(membership) => {
+            let stored = openraft::StoredMembership::new(Some(entry.log_id), membership.clone());
+            let checked = crate::CaptureMembership::try_from(&stored)?;
+            (2, checked.encode()?)
+        }
+    };
+    let mut payload = b"CENR\x01".to_vec();
+    payload.push(kind);
+    payload.extend(
+        u64::try_from(body.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes(),
+    );
+    payload.extend(body);
+    encode_log(&crate::LogEntry { id, payload })
+}
+
+pub(crate) fn decode_native_entry(entry: crate::LogEntry) -> Result<crate::CaptureRaftEntry, StoreError> {
+    let bytes = &entry.payload;
+    if bytes.len() < 14 || &bytes[..5] != b"CENR\x01" {
+        return Err(StoreError::CorruptRecord);
+    }
+    let length = usize::try_from(number(&bytes[6..14])?).map_err(|_| StoreError::CorruptRecord)?;
+    let body = &bytes[14..];
+    if length != body.len() {
+        return Err(StoreError::CorruptRecord);
+    }
+    let payload = match bytes[5] {
+        0 if body.is_empty() => openraft::EntryPayload::Blank,
+        1 => openraft::EntryPayload::Normal(crate::CaptureCommand::try_created(DecideCreated::decode(body)?)?),
+        2 => {
+            let membership = crate::CaptureMembership::decode(body).map_err(|_| StoreError::CorruptRecord)?;
+            if *membership.log_id() != Some(entry.id) {
+                return Err(StoreError::CorruptRecord);
+            }
+            let stored = openraft::StoredMembership::try_from(&membership).map_err(|_| StoreError::CorruptRecord)?;
+            openraft::EntryPayload::Membership(stored.membership().clone())
+        }
+        _ => return Err(StoreError::CorruptRecord),
+    };
+    Ok(crate::CaptureRaftEntry {
+        log_id: entry.id.into(),
+        payload,
+    })
+}
+
+fn apply_created(
+    tx: &redb::WriteTransaction,
+    id: LogId,
+    command: &DecideCreated,
+    canonical: &[u8],
+    binding: &CaptureBinding,
+) -> Result<CreatedResult, StoreError> {
+    let mut operations = tx.open_table(OPERATIONS)?;
+    let existing = operations
+        .get(command.operation_id.as_slice())?
+        .map(|v| decode_operation(v.value()))
+        .transpose()?;
+    let result = if let Some(existing) = existing {
+        if existing.canonical_command == canonical {
+            existing.result
+        } else {
+            CreatedResult::OperationConflict
+        }
+    } else {
+        let mut heads = tx.open_table(HEADS)?;
+        let mut versions = tx.open_table(VERSIONS)?;
+        let object = object_bytes(&command.object)?;
+        let key = version_key(&command.object, command.prepared.version_id)?;
+        let actual = heads
+            .get(object.as_slice())?
+            .map(|v| {
+                let mut d = Decoder::new(v.value());
+                let id = d.uuid()?;
+                d.finish()?;
+                Ok::<_, StoreError>(id)
+            })
+            .transpose()?;
+        let outcome = if command.binding_revision != binding.revision {
+            CreatedResult::BindingMismatch
+        } else if actual != command.expected_head {
+            CreatedResult::HeadMismatch { actual_head: actual }
+        } else if versions.get(key.as_slice())?.is_some() {
+            CreatedResult::VersionConflict
+        } else {
+            let mut version = Vec::new();
+            put_prepared(&mut version, &command.prepared);
+            version.extend(encode_id(id));
+            version.extend(binding.revision.to_be_bytes());
+            versions.insert(key.as_slice(), version.as_slice())?;
+            let mut event = object;
+            put_prepared(&mut event, &command.prepared);
+            event.extend(encode_id(id));
+            event.extend(binding_bytes(&binding)?);
+            tx.open_table(EVENTS)?
+                .insert(command.operation_id.as_slice(), event.as_slice())?;
+            heads.insert(object_bytes(&command.object)?.as_slice(), command.prepared.version_id.as_slice())?;
+            CreatedResult::Created {
+                version_id: command.prepared.version_id,
+                event_id: command.operation_id,
+                decision_log_id: id,
+            }
+        };
+        let mut op = u64::try_from(canonical.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes()
+            .to_vec();
+        op.extend(canonical);
+        op.extend(result_bytes(&outcome));
+        operations.insert(command.operation_id.as_slice(), op.as_slice())?;
+        outcome
+    };
+    Ok(result)
+}
+
+fn validate_response_kind(entry: &crate::CaptureRaftEntry, response: &crate::CaptureResponse) -> Result<(), StoreError> {
+    let kind = match entry.payload {
+        openraft::EntryPayload::Blank => crate::CaptureResponseKind::Blank,
+        openraft::EntryPayload::Normal(_) => crate::CaptureResponseKind::Created,
+        openraft::EntryPayload::Membership(_) => crate::CaptureResponseKind::Membership,
+    };
+    if response.kind() != kind || *response.log_id() != LogId::from(entry.log_id) {
+        return Err(StoreError::CorruptRecord);
+    }
+    Ok(())
+}
+fn encode_response(response: &crate::CaptureResponse) -> Result<Vec<u8>, StoreError> {
+    let (kind, body) = match response.kind() {
+        crate::CaptureResponseKind::Blank => (0, vec![]),
+        crate::CaptureResponseKind::Created => (1, result_bytes(response.created_result().ok_or(StoreError::CorruptRecord)?)),
+        crate::CaptureResponseKind::Membership => (2, vec![]),
+    };
+    let mut bytes = b"CRSP\x01".to_vec();
+    bytes.push(kind);
+    bytes.extend(encode_id(*response.log_id()));
+    bytes.extend(
+        u64::try_from(body.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes(),
+    );
+    bytes.extend(body);
+    Ok(bytes)
+}
+fn decode_response(bytes: &[u8]) -> Result<crate::CaptureResponse, StoreError> {
+    if bytes.len() < 38
+        || &bytes[..5] != b"CRSP\x01"
+        || number(&bytes[30..38])? != u64::try_from(bytes.len() - 38).map_err(|_| StoreError::CorruptRecord)?
+    {
+        return Err(StoreError::CorruptRecord);
+    }
+    let id = crate::decode_id(&bytes[6..30])?;
+    match bytes[5] {
+        0 if bytes.len() == 38 => crate::CaptureResponse::blank(id),
+        1 => {
+            let mut body = encode_id(id);
+            body.extend(&bytes[38..]);
+            crate::CaptureResponse::created(decode_apply(&body)?)
+        }
+        2 if bytes.len() == 38 => crate::CaptureResponse::membership(id),
+        _ => Err(StoreError::CorruptRecord),
+    }
+}
+fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated, StoreError> {
+    if native {
+        match decode_native_entry(entry.clone())?.payload {
+            openraft::EntryPayload::Normal(command) => Ok(command.created().clone()),
+            _ => Err(StoreError::CorruptRecord),
+        }
+    } else {
+        DecideCreated::decode(&entry.payload)
+    }
+}
+
+pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
+    let meta = tx.open_table(META)?;
+    if number(required(meta.get("schema")?)?.value())? != 4 {
+        return Err(StoreError::UnsupportedSchema);
+    }
+    let last = read_applied(&meta)?;
+    let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
+        .map_err(|_| StoreError::CorruptRecord)?;
+    if membership
+        .log_id()
+        .is_some_and(|id| last.is_none_or(|last| id.index > last.index))
+    {
+        return Err(StoreError::CorruptRecord);
+    }
+    Ok((last, membership))
+}
+
+impl Store {
+    /// Explicitly opt an empty, bound store into native entry persistence.
+    pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 3 | 4) {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            read_binding(&meta)?;
+            if schema == 4 {
+                let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
+                    .map_err(|_| StoreError::CorruptRecord)?;
+                let last = read_applied(&meta)?;
+                if membership
+                    .log_id()
+                    .is_some_and(|id| last.is_none_or(|last| id.index > last.index))
+                    || (last.is_none() && (!membership.configs().is_empty() || !membership.nodes().is_empty()))
+                {
+                    return Err(StoreError::CorruptRecord);
+                }
+                return Ok(());
+            }
+            if read_applied(&meta)?.is_some() || !tx.open_table(LOGS)?.is_empty()? {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
+                if !tx.open_table(table)?.is_empty()? {
+                    return Err(StoreError::UnsupportedSchema);
+                }
+            }
+            if !tx.open_table(APPLY_RESULTS)?.is_empty()? || meta.get("committed")?.is_some() || meta.get("tail")?.is_some() {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            if meta.get("native_membership")?.is_some() {
+                return Err(StoreError::CorruptRecord);
+            }
+            let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
+            meta.insert("native_membership", membership.encode()?.as_slice())?;
+            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
+        }
+        self.commit(tx)
+    }
+    pub fn append_native(&mut self, entries: &[crate::CaptureRaftEntry]) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            if number(required(meta.get("schema")?)?.value())? != 4 {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            if entries.is_empty() {
+                return Ok(());
+            }
+            let tail = meta.get("tail")?.map(|value| crate::decode_id(value.value())).transpose()?;
+            let mut next = next_index(tail.map(|id| id.index))?;
+            let mut encoded = Vec::with_capacity(entries.len());
+            for (position, entry) in entries.iter().enumerate() {
+                if entry.log_id.index != next {
+                    return Err(StoreError::InvalidLogSequence);
+                }
+                encoded.push(encode_native_entry(entry)?);
+                if position + 1 < entries.len() {
+                    next = next_index(Some(next))?;
+                }
+            }
+            let mut logs = tx.open_table(LOGS)?;
+            for (entry, bytes) in entries.iter().zip(&encoded) {
+                logs.insert(entry.log_id.index, bytes.as_slice())?;
+            }
+            let last = entries.last().ok_or(StoreError::InvalidLogSequence)?;
+            meta.insert("tail", encode_id(last.log_id.into()).as_slice())?;
+        }
+        self.commit(tx)
+    }
+    pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
+        let tx = self.database.begin_read()?;
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        tx.open_table(LOGS)?
+            .get(index)?
+            .map(|value| {
+                let entry = decode_log(value.value())?;
+                if entry.id.index != index {
+                    return Err(StoreError::CorruptRecord);
+                }
+                decode_native_entry(entry)
+            })
+            .transpose()
+    }
+    pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
+        let tx = self.database.begin_read()?;
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        if start > end {
+            return Err(StoreError::InvalidLogSequence);
+        }
+        let logs = tx.open_table(LOGS)?;
+        let mut entries = Vec::new();
+        for row in logs.range(start..end)? {
+            let (key, value) = row?;
+            let entry = decode_log(value.value())?;
+            if entry.id.index != key.value() {
+                return Err(StoreError::CorruptRecord);
+            }
+            entries.push(decode_native_entry(entry)?);
+        }
+        Ok(entries)
+    }
+    pub fn apply_next_native(&mut self, id: LogId) -> Result<crate::CaptureResponse, StoreError> {
+        let tx = self.write_transaction()?;
+        let response;
+        {
+            let mut meta = tx.open_table(META)?;
+            if number(required(meta.get("schema")?)?.value())? != 4 {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            let binding = read_binding(&meta)?;
+            let logs = tx.open_table(LOGS)?;
+            let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
+            let entry = decode_log(required(logs.get(id.index)?)?.value())?;
+            if entry.id != id || id.index > committed.index {
+                return Err(StoreError::InvalidCommittedLog);
+            }
+            let native = decode_native_entry(entry)?;
+            let last = read_applied(&meta)?;
+            let mut results = tx.open_table(APPLY_RESULTS)?;
+            if last.is_some_and(|last| id.index <= last.index) {
+                let saved = decode_response(required(results.get(id.index)?)?.value())?;
+                if *saved.log_id() != id {
+                    return Err(StoreError::InvalidApplySequence);
+                }
+                validate_response_kind(&native, &saved)?;
+                return Ok(saved);
+            }
+            if id.index != next_index(last.map(|last| last.index))? {
+                return Err(StoreError::InvalidApplySequence);
+            }
+            response = match native.payload {
+                openraft::EntryPayload::Blank => crate::CaptureResponse::blank(id)?,
+                openraft::EntryPayload::Normal(command) => {
+                    let canonical = command.created().encode()?;
+                    let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
+                    crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
+                }
+                openraft::EntryPayload::Membership(membership) => {
+                    let stored = openraft::StoredMembership::new(Some(id.into()), membership);
+                    let checked = crate::CaptureMembership::try_from(&stored)?;
+                    meta.insert("native_membership", checked.encode()?.as_slice())?;
+                    crate::CaptureResponse::membership(id)?
+                }
+            };
+            results.insert(id.index, encode_response(&response)?.as_slice())?;
+            let mut last = vec![1];
+            last.extend(encode_id(id));
+            meta.insert("last_applied", last.as_slice())?;
+        }
+        self.commit(tx)?;
+        Ok(response)
+    }
+    pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
+        let tx = self.database.begin_read()?;
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        let results = tx.open_table(APPLY_RESULTS)?;
+        results.get(index)?.map(|v| decode_response(v.value())).transpose()
+    }
+    pub fn native_application_state(&self) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
+        native_application_state(&self.database.begin_read()?)
+    }
+    /// Bootstrap the immutable capture policy. The caller must be the trusted
+    /// deployment provisioning owner. This crate does not authenticate callers;
+    /// do not expose this public method through an ordinary object request API.
+    /// Prepared identities assert identity only, not physical data durability.
+    pub fn pin_capture_binding(&mut self, binding: CaptureBinding) -> Result<(), StoreError> {
+        let encoded = binding_bytes(&binding)?;
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+                if read_binding(&meta)? == binding {
+                    return Ok(());
+                }
+                return Err(StoreError::BindingConflict);
+            }
+            if meta.get("capture_binding")?.is_some() || meta.get("last_applied")?.is_some() {
+                return Err(StoreError::CorruptRecord);
+            }
+            for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
+                if !tx.open_table(table)?.is_empty()? {
+                    return Err(StoreError::CorruptRecord);
+                }
+            }
+            if !tx.open_table(APPLY_RESULTS)?.is_empty()? {
+                return Err(StoreError::CorruptRecord);
+            }
+            meta.insert("capture_binding", encoded.as_slice())?;
+            meta.insert("last_applied", &[0][..])?;
+            meta.insert("schema", 3u64.to_be_bytes().as_slice())?;
+        }
+        self.commit(tx)
+    }
+    /// Apply the next exact persisted committed Created command atomically.
+    pub fn apply_next_created(&mut self, id: LogId) -> Result<CreatedResult, StoreError> {
+        let tx = self.write_transaction()?;
+        let result;
+        {
+            let mut meta = tx.open_table(META)?;
+            if number(required(meta.get("schema")?)?.value())? != 3 {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            let binding = read_binding(&meta)?;
+            let logs = tx.open_table(LOGS)?;
+            let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
+            let entry = decode_log(required(logs.get(id.index)?)?.value())?;
+            if entry.id != id || id.index > committed.index {
+                return Err(StoreError::InvalidCommittedLog);
+            }
+            let last = read_applied(&meta)?;
+            let mut results = tx.open_table(APPLY_RESULTS)?;
+            if last.is_some_and(|last| id.index <= last.index) {
+                let saved = decode_apply(required(results.get(id.index)?)?.value())?;
+                if saved.log_id != id {
+                    return Err(StoreError::InvalidApplySequence);
+                }
+                return Ok(saved.result);
+            }
+            if id.index != next_index(last.map(|last| last.index))? {
+                return Err(StoreError::InvalidApplySequence);
+            }
+            let command = DecideCreated::decode(&entry.payload)?;
+            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
+            let mut applied = encode_id(id);
+            applied.extend(result_bytes(&result));
+            results.insert(id.index, applied.as_slice())?;
+            let mut last = vec![1];
+            last.extend(encode_id(id));
+            meta.insert("last_applied", last.as_slice())?;
+        }
+        self.commit(tx)?;
+        Ok(result)
+    }
+    pub fn capture_binding(&self) -> Result<Option<CaptureBinding>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let meta = tx.open_table(META)?;
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+            return Ok(None);
+        }
+        Ok(Some(read_binding(&meta)?))
+    }
+    pub fn last_applied(&self) -> Result<Option<LogId>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let meta = tx.open_table(META)?;
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+            return Ok(None);
+        }
+        read_applied(&meta)
+    }
+    pub fn created_head(&self, object: &ObjectIdentity) -> Result<Option<[u8; 16]>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let t = tx.open_table(HEADS)?;
+        t.get(object_bytes(object)?.as_slice())?
+            .map(|v| {
+                let mut d = Decoder::new(v.value());
+                let id = d.uuid()?;
+                d.finish()?;
+                Ok(id)
+            })
+            .transpose()
+    }
+    pub fn created_version(&self, object: &ObjectIdentity, version: [u8; 16]) -> Result<Option<CreatedVersion>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let t = tx.open_table(VERSIONS)?;
+        t.get(version_key(object, version)?.as_slice())?
+            .map(|v| decode_version(v.value()))
+            .transpose()
+    }
+    pub fn created_event(&self, operation: [u8; 16]) -> Result<Option<CreatedEvent>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let t = tx.open_table(EVENTS)?;
+        t.get(operation.as_slice())?.map(|v| decode_event(v.value())).transpose()
+    }
+    pub fn created_operation(&self, operation: [u8; 16]) -> Result<Option<CreatedOperation>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let t = tx.open_table(OPERATIONS)?;
+        t.get(operation.as_slice())?.map(|v| decode_operation(v.value())).transpose()
+    }
+    pub fn created_apply_result(&self, index: u64) -> Result<Option<CreatedApplyResult>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let t = tx.open_table(APPLY_RESULTS)?;
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
+            t.get(index)?
+                .map(|v| {
+                    decode_response(v.value()).map(|r| {
+                        r.created_result().cloned().map(|result| CreatedApplyResult {
+                            log_id: *r.log_id(),
+                            result,
+                        })
+                    })
+                })
+                .transpose()
+                .map(Option::flatten)
+        } else {
+            t.get(index)?.map(|v| decode_apply(v.value())).transpose()
+        }
+    }
+}
+
+pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
+    let meta = tx.open_table(META)?;
+    let schema = number(required(meta.get("schema")?)?.value())?;
+    let native = schema == 4;
+    if !matches!(schema, 3 | 4) {
+        return Ok(());
+    }
+    let binding = read_binding(&meta).map_err(|error| match error {
+        StoreError::BindingNotPinned => StoreError::CorruptRecord,
+        other => other,
+    })?;
+    let last = read_applied(&meta)?;
+    let logs = tx.open_table(LOGS)?;
+    let committed = validate_committed(&meta, &logs)?;
+    let heads = tx.open_table(HEADS)?;
+    let versions = tx.open_table(VERSIONS)?;
+    let events = tx.open_table(EVENTS)?;
+    let operations = tx.open_table(OPERATIONS)?;
+    let results = tx.open_table(APPLY_RESULTS)?;
+    if last.is_some_and(|last| committed.is_none_or(|c| last.index > c.index)) {
+        return Err(StoreError::CorruptRecord);
+    }
+    let mut next = Some(0);
+    let mut observed = None;
+    let mut first_operations = BTreeMap::new();
+    let mut expected_membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
+    let mut replay_operations = BTreeMap::<[u8; 16], (Vec<u8>, CreatedResult)>::new();
+    let mut replay_heads = BTreeMap::<Vec<u8>, [u8; 16]>::new();
+    let mut replay_versions = std::collections::BTreeSet::new();
+    for row in results.iter()? {
+        let (k, v) = row?;
+        let entry = decode_log(required(logs.get(k.value())?)?.value())?;
+        let r = if native {
+            let response = decode_response(v.value())?;
+            let native_entry = decode_native_entry(entry.clone())?;
+            validate_response_kind(&native_entry, &response)?;
+            if next != Some(k.value()) || response.log_id().index != k.value() {
+                return Err(StoreError::CorruptRecord);
+            }
+            observed = Some(*response.log_id());
+            next = k.value().checked_add(1);
+            match native_entry.payload {
+                openraft::EntryPayload::Membership(membership) => {
+                    expected_membership =
+                        crate::CaptureMembership::try_from(&openraft::StoredMembership::new(Some(entry.id.into()), membership))?;
+                    continue;
+                }
+                openraft::EntryPayload::Blank => continue,
+                openraft::EntryPayload::Normal(_) => CreatedApplyResult {
+                    log_id: *response.log_id(),
+                    result: response.created_result().cloned().ok_or(StoreError::CorruptRecord)?,
+                },
+            }
+        } else {
+            decode_apply(v.value())?
+        };
+        if !native && (next != Some(k.value()) || r.log_id.index != k.value()) {
+            return Err(StoreError::CorruptRecord);
+        }
+        if entry.id != r.log_id {
+            return Err(StoreError::CorruptRecord);
+        }
+        let c = entry_command(&entry, native)?;
+        let canonical = c.encode()?;
+        let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
+        let canonical_matches = op.canonical_command == canonical;
+        let expected = if canonical_matches {
+            op.result
+        } else {
+            CreatedResult::OperationConflict
+        };
+        if native {
+            let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
+                if *previous == canonical {
+                    result.clone()
+                } else {
+                    CreatedResult::OperationConflict
+                }
+            } else {
+                let object = object_bytes(&c.object)?;
+                let key = version_key(&c.object, c.prepared.version_id)?;
+                let actual = replay_heads.get(&object).copied();
+                let result = if c.binding_revision != binding.revision {
+                    CreatedResult::BindingMismatch
+                } else if actual != c.expected_head {
+                    CreatedResult::HeadMismatch { actual_head: actual }
+                } else if replay_versions.contains(&key) {
+                    CreatedResult::VersionConflict
+                } else {
+                    replay_heads.insert(object, c.prepared.version_id);
+                    replay_versions.insert(key);
+                    CreatedResult::Created {
+                        version_id: c.prepared.version_id,
+                        event_id: c.operation_id,
+                        decision_log_id: r.log_id,
+                    }
+                };
+                replay_operations.insert(c.operation_id, (canonical.clone(), result.clone()));
+                result
+            };
+            if replay != r.result {
+                return Err(StoreError::CorruptRecord);
+            }
+        }
+        if expected != r.result {
+            return Err(StoreError::CorruptRecord);
+        }
+        if canonical_matches {
+            first_operations.entry(c.operation_id).or_insert(r.log_id);
+        }
+        observed = Some(r.log_id);
+        next = k.value().checked_add(1)
+    }
+    if observed != last {
+        return Err(StoreError::CorruptRecord);
+    }
+    if native {
+        if native_application_state(tx)?.1 != expected_membership
+            || heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
+            || versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
+        {
+            return Err(StoreError::CorruptRecord);
+        }
+        for (object, id) in replay_heads {
+            if required(heads.get(object.as_slice())?)?.value() != id {
+                return Err(StoreError::CorruptRecord);
+            }
+        }
+    }
+    for row in operations.iter()? {
+        let (k, v) = row?;
+        let op = decode_operation(v.value())?;
+        let c = DecideCreated::decode(&op.canonical_command)?;
+        if k.value() != c.operation_id {
+            return Err(StoreError::CorruptRecord);
+        }
+        let first = required(first_operations.get(&c.operation_id))?;
+        if let CreatedResult::Created {
+            version_id,
+            event_id,
+            decision_log_id,
+        } = op.result
+        {
+            if version_id != c.prepared.version_id
+                || event_id != c.operation_id
+                || c.binding_revision != binding.revision
+                || decision_log_id != *first
+            {
+                return Err(StoreError::CorruptRecord);
+            }
+            let event = decode_event(required(events.get(event_id.as_slice())?)?.value())?;
+            if event
+                != (CreatedEvent {
+                    object: c.object.clone(),
+                    prepared: c.prepared.clone(),
+                    decision_log_id,
+                    binding: binding.clone(),
+                })
+            {
+                return Err(StoreError::CorruptRecord);
+            }
+        } else if events.get(c.operation_id.as_slice())?.is_some() {
+            return Err(StoreError::CorruptRecord);
+        }
+    }
+    for row in events.iter()? {
+        let (k, v) = row?;
+        let e = decode_event(v.value())?;
+        if e.binding != binding {
+            return Err(StoreError::CorruptRecord);
+        }
+        let op = decode_operation(required(operations.get(k.value())?)?.value())?;
+        let c = DecideCreated::decode(&op.canonical_command)?;
+        if c.object != e.object
+            || c.prepared != e.prepared
+            || op.result
+                != (CreatedResult::Created {
+                    version_id: e.prepared.version_id,
+                    event_id: c.operation_id,
+                    decision_log_id: e.decision_log_id,
+                })
+        {
+            return Err(StoreError::CorruptRecord);
+        }
+        let version =
+            decode_version(required(versions.get(version_key(&e.object, e.prepared.version_id)?.as_slice())?)?.value())?;
+        if version
+            != (CreatedVersion {
+                prepared: e.prepared.clone(),
+                decision_log_id: e.decision_log_id,
+                binding_revision: binding.revision,
+            })
+        {
+            return Err(StoreError::CorruptRecord);
+        }
+    }
+    for row in versions.iter()? {
+        let (k, v) = row?;
+        let version = decode_version(v.value())?;
+        let mut d = Decoder::new(k.value());
+        let object = d.object()?;
+        let id = d.uuid()?;
+        d.finish()?;
+        if id != version.prepared.version_id || version.binding_revision != binding.revision {
+            return Err(StoreError::CorruptRecord);
+        }
+        let entry = decode_log(required(logs.get(version.decision_log_id.index)?)?.value())?;
+        let c = entry_command(&entry, native)?;
+        if entry.id != version.decision_log_id
+            || c.object != object
+            || c.prepared != version.prepared
+            || events.get(c.operation_id.as_slice())?.is_none()
+        {
+            return Err(StoreError::CorruptRecord);
+        }
+        let head = required(heads.get(object_bytes(&object)?.as_slice())?)?;
+        let mut d = Decoder::new(head.value());
+        let head_id = d.uuid()?;
+        d.finish()?;
+        let current = decode_version(required(versions.get(version_key(&object, head_id)?.as_slice())?)?.value())?;
+        if current.decision_log_id.index < version.decision_log_id.index {
+            return Err(StoreError::CorruptRecord);
+        }
+    }
+    for row in heads.iter()? {
+        let (k, v) = row?;
+        let mut d = Decoder::new(k.value());
+        let object = d.object()?;
+        d.finish()?;
+        let mut d = Decoder::new(v.value());
+        let id = d.uuid()?;
+        d.finish()?;
+        decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
+    }
+    Ok(())
+}

crates/capture-authority/src/application_tests.rs

@@ -0,0 +1,531 @@
+use super::*;
+use crate::application::{APPLY_RESULTS, EVENTS, OPERATIONS, VERSIONS};
+use crate::tests::CrashBackend;
+use redb::ReadableTableMetadata;
+use std::{
+    io::{BufRead, BufReader, Write},
+    process::{Command, Stdio},
+    sync::mpsc,
+    time::Duration,
+};
+fn identity() -> NodeIdentity {
+    NodeIdentity {
+        cluster_id: [7; 16],
+        node_id: 0,
+    }
+}
+fn binding() -> CaptureBinding {
+    CaptureBinding {
+        revision: 9,
+        binding_id: [8; 16],
+        target_id: [9; 16],
+    }
+}
+fn command() -> DecideCreated {
+    DecideCreated {
+        operation_id: [1; 16],
+        object: ObjectIdentity {
+            bucket_incarnation: [2; 16],
+            key: vec![0, 255, 3],
+        },
+        expected_head: None,
+        prepared: PreparedIdentity {
+            version_id: [3; 16],
+            preparation_id: [4; 16],
+            content_digest: [5; 32],
+            content_length: 123,
+        },
+        binding_revision: 9,
+    }
+}
+fn id(index: u64) -> LogId {
+    LogId {
+        term: 8,
+        leader_node: 0,
+        index,
+    }
+}
+fn append(store: &mut Store, index: u64, c: &DecideCreated) {
+    store
+        .append(&[LogEntry {
+            id: id(index),
+            payload: c.encode().expect("encode command"),
+        }])
+        .expect("append command");
+}
+fn apply(store: &mut Store, index: u64, c: &DecideCreated) -> CreatedResult {
+    append(store, index, c);
+    store.save_committed(id(index)).expect("commit log");
+    store.apply_next_created(id(index)).expect("apply command")
+}
+fn success(c: &DecideCreated, index: u64) -> CreatedResult {
+    CreatedResult::Created {
+        version_id: c.prepared.version_id,
+        event_id: c.operation_id,
+        decision_log_id: id(index),
+    }
+}
+fn complete(store: &Store) {
+    let c = command();
+    assert_eq!(store.capture_binding().expect("binding"), Some(binding()));
+    assert_eq!(
+        store.created_head(&c.object).expect("head"),
+        Some(c.prepared.version_id),
+        "U11 acknowledged head lost"
+    );
+    assert_eq!(
+        store.created_version(&c.object, c.prepared.version_id).expect("version"),
+        Some(CreatedVersion {
+            prepared: c.prepared.clone(),
+            decision_log_id: id(0),
+            binding_revision: 9
+        })
+    );
+    assert_eq!(
+        store.created_event(c.operation_id).expect("event"),
+        Some(CreatedEvent {
+            object: c.object.clone(),
+            prepared: c.prepared.clone(),
+            decision_log_id: id(0),
+            binding: binding()
+        }),
+        "U11 complete event missing"
+    );
+    assert_eq!(
+        store.created_operation(c.operation_id).expect("operation"),
+        Some(CreatedOperation {
+            canonical_command: c.encode().expect("encode"),
+            result: success(&c, 0)
+        })
+    );
+    assert_eq!(
+        store.created_apply_result(0).expect("result"),
+        Some(CreatedApplyResult {
+            log_id: id(0),
+            result: success(&c, 0)
+        })
+    );
+    assert_eq!(store.last_applied().expect("last applied"), Some(id(0)));
+}
+fn fixture() -> (tempfile::TempDir, Store) {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let mut store = Store::initialize(dir.path().join("db"), identity()).expect("initialize");
+    store.pin_capture_binding(binding()).expect("pin");
+    (dir, store)
+}
+fn reopen(path: &std::path::Path) -> Store {
+    let result = Store::open(path, identity());
+    assert!(result.is_ok(), "U11 complete reopen: {:?}", result.as_ref().err());
+    result.expect("asserted successful reopen")
+}
+#[test]
+fn created_codec_is_canonical_and_strict() {
+    let c = command();
+    let b = c.encode().expect("encode");
+    let mut golden = b"CRTD\x01".to_vec();
+    golden.extend([1; 16]);
+    golden.extend([2; 16]);
+    golden.extend(3u32.to_be_bytes());
+    golden.extend([0, 255, 3]);
+    golden.push(0);
+    golden.extend([3; 16]);
+    golden.extend([4; 16]);
+    golden.extend([5; 32]);
+    golden.extend(123u64.to_be_bytes());
+    golden.extend(9u64.to_be_bytes());
+    assert_eq!(b, golden);
+    assert_eq!(DecideCreated::decode(&b).expect("decode"), c);
+    let mut variants = Vec::new();
+    let mut v = c.clone();
+    v.operation_id = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.object.bucket_incarnation = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.object.key.push(4);
+    variants.push(v);
+    let mut v = c.clone();
+    v.expected_head = Some([6; 16]);
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.version_id = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.preparation_id = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.content_digest = [6; 32];
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.content_length += 1;
+    variants.push(v);
+    let mut v = c.clone();
+    v.binding_revision += 1;
+    variants.push(v);
+    for v in variants {
+        assert_ne!(v.encode().expect("variant"), b)
+    }
+    for n in 0..b.len() {
+        assert!(DecideCreated::decode(&b[..n]).is_err())
+    }
+    let mut bad = b.clone();
+    bad[4] = 2;
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = b.clone();
+    bad[44] = 2;
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = b.clone();
+    bad.push(0);
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = b;
+    bad[37..41].copy_from_slice(&4097u32.to_be_bytes());
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = c.clone();
+    bad.operation_id = [0; 16];
+    assert!(bad.encode().is_err());
+    bad = c;
+    bad.object.key = vec![0; MAX_OBJECT_KEY_LENGTH + 1];
+    assert!(bad.encode().is_err());
+}
+#[test]
+fn created_committed_order_and_identity_are_enforced() {
+    let (_dir, mut s) = fixture();
+    let c = command();
+    append(&mut s, 0, &c);
+    append(&mut s, 1, &c);
+    assert!(s.apply_next_created(id(0)).is_err());
+    assert_eq!(s.last_applied().expect("no apply"), None);
+    s.save_committed(id(1)).expect("committed");
+    assert!(matches!(s.apply_next_created(id(1)), Err(StoreError::InvalidApplySequence)));
+    let mut wrong = id(0);
+    wrong.term += 1;
+    assert!(s.apply_next_created(wrong).is_err());
+    wrong = id(0);
+    wrong.leader_node += 1;
+    assert!(s.apply_next_created(wrong).is_err());
+    assert_eq!(s.apply_next_created(id(0)).expect("first"), success(&c, 0));
+    complete(&s);
+    assert_eq!(s.apply_next_created(id(0)).expect("retry"), success(&c, 0));
+    complete(&s);
+    assert!(s.apply_next_created(wrong).is_err());
+    assert_eq!(s.apply_next_created(id(1)).expect("duplicate log"), success(&c, 0));
+    assert_eq!(s.last_applied().expect("last"), Some(id(1)));
+    assert_eq!(
+        s.created_apply_result(1).expect("result"),
+        Some(CreatedApplyResult {
+            log_id: id(1),
+            result: success(&c, 0)
+        })
+    );
+    let tx = s.database.begin_read().expect("read");
+    assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 1);
+}
+#[test]
+fn created_head_version_event_and_dedup_reopen_together() {
+    let (dir, mut s) = fixture();
+    let c = command();
+    assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
+    drop(s);
+    let mut s = reopen(&dir.path().join("db"));
+    complete(&s);
+    assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
+    let mut changed = c.clone();
+    changed.prepared.content_digest = [6; 32];
+    assert_eq!(
+        apply(&mut s, 2, &changed),
+        CreatedResult::OperationConflict,
+        "U11 full canonical operation conflict"
+    );
+    assert_eq!(
+        s.created_operation(c.operation_id)
+            .expect("original")
+            .expect("operation")
+            .result,
+        success(&c, 0)
+    );
+    let mut stale = c.clone();
+    stale.operation_id = [6; 16];
+    stale.prepared.version_id = [6; 16];
+    assert_eq!(
+        apply(&mut s, 3, &stale),
+        CreatedResult::HeadMismatch {
+            actual_head: Some(c.prepared.version_id)
+        }
+    );
+    let mut reuse = c.clone();
+    reuse.operation_id = [7; 16];
+    reuse.expected_head = Some(c.prepared.version_id);
+    assert_eq!(apply(&mut s, 4, &reuse), CreatedResult::VersionConflict);
+    let mut replacement = stale.clone();
+    replacement.operation_id = [10; 16];
+    replacement.expected_head = Some(c.prepared.version_id);
+    assert_eq!(apply(&mut s, 5, &replacement), success(&replacement, 5));
+    assert_eq!(s.created_head(&c.object).expect("head"), Some(replacement.prepared.version_id));
+    assert_eq!(
+        s.created_version(&c.object, c.prepared.version_id)
+            .expect("old version")
+            .expect("version")
+            .decision_log_id,
+        id(0)
+    );
+    assert!(s.created_event(c.operation_id).expect("old event").is_some());
+    assert!(s.created_event(replacement.operation_id).expect("new event").is_some());
+    assert!(s.created_event(stale.operation_id).expect("rejection event").is_none());
+    assert_eq!(
+        apply(&mut s, 6, &stale),
+        CreatedResult::HeadMismatch {
+            actual_head: Some(c.prepared.version_id)
+        }
+    );
+    drop(s);
+    let s = reopen(&dir.path().join("db"));
+    assert_eq!(s.last_applied().expect("last"), Some(id(6)));
+    let tx = s.database.begin_read().expect("read");
+    assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 2);
+}
+#[test]
+fn created_binding_is_pinned_and_not_caller_selected() {
+    let (dir, mut s) = fixture();
+    let mut bad = command();
+    bad.binding_revision = 10;
+    assert_eq!(apply(&mut s, 0, &bad), CreatedResult::BindingMismatch);
+    assert!(s.created_head(&bad.object).expect("head").is_none());
+    assert!(s.created_event(bad.operation_id).expect("event").is_none());
+    assert_eq!(
+        s.created_operation(bad.operation_id)
+            .expect("operation")
+            .expect("saved")
+            .result,
+        CreatedResult::BindingMismatch
+    );
+    s.pin_capture_binding(binding()).expect("idempotent pin");
+    let mut changed = binding();
+    changed.target_id = [10; 16];
+    assert!(matches!(s.pin_capture_binding(changed), Err(StoreError::BindingConflict)));
+    drop(s);
+    let s = reopen(&dir.path().join("db"));
+    assert_eq!(s.capture_binding().expect("binding"), Some(binding()));
+}
+#[test]
+fn created_schema_compatibility_and_corruption_are_checked() {
+    let dir = tempfile::tempdir().expect("dir");
+    let path = dir.path().join("db");
+    let s = Store::initialize(&path, identity()).expect("schema1");
+    assert_eq!(s.capture_binding().expect("unmigrated"), None);
+    drop(s);
+    let mut s = reopen(&path);
+    assert_eq!(s.last_applied().expect("no apply"), None);
+    append(&mut s, 0, &command());
+    s.save_committed(id(0)).expect("schema2");
+    drop(s);
+    let mut s = reopen(&path);
+    assert_eq!(s.capture_binding().expect("schema2 unchanged"), None);
+    s.pin_capture_binding(binding()).expect("explicit upgrade");
+    drop(s);
+    let mut s = reopen(&path);
+    assert_eq!(s.capture_binding().expect("schema3"), Some(binding()));
+    s.apply_next_created(id(0)).expect("apply");
+    drop(s);
+    let db = Database::open(&path).expect("raw db");
+    let tx = db.begin_write().expect("write corruption");
+    tx.open_table(EVENTS)
+        .expect("events")
+        .remove(command().operation_id.as_slice())
+        .expect("remove reference");
+    tx.commit().expect("commit corruption");
+    drop(db);
+    let result = Store::open(&path, identity());
+    assert!(
+        matches!(result, Err(StoreError::CorruptRecord)),
+        "U11 missing event rejected before expect"
+    );
+    let (dir, mut s) = fixture();
+    assert_eq!(s.read_committed().expect("empty committed"), None);
+    append(&mut s, 0, &command());
+    s.save_committed(id(0)).expect("committed schema3");
+    assert_eq!(s.capture_binding().expect("no downgrade"), Some(binding()));
+    drop(s);
+    let path = dir.path().join("db");
+    let db = Database::open(&path).expect("raw db");
+    let tx = db.begin_write().expect("write");
+    let mut marker = vec![1];
+    marker.extend(encode_id(id(0)));
+    tx.open_table(META)
+        .expect("meta")
+        .insert("last_applied", marker.as_slice())
+        .expect("corrupt applied");
+    tx.commit().expect("commit");
+    drop(db);
+    assert!(
+        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
+        "U11 last applied missing result rejected"
+    );
+    let (dir, s) = fixture();
+    drop(s);
+    let path = dir.path().join("db");
+    let db = Database::open(&path).expect("raw pinned db");
+    let tx = db.begin_write().expect("missing binding write");
+    tx.open_table(META)
+        .expect("meta")
+        .remove("capture_binding")
+        .expect("remove binding");
+    tx.commit().expect("binding corruption commit");
+    drop(db);
+    assert!(
+        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
+        "U11 missing pinned binding rejected"
+    );
+
+    let (dir, mut s) = fixture();
+    let c = command();
+    assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
+    assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
+    drop(s);
+    let path = dir.path().join("db");
+    let db = Database::open(&path).expect("raw duplicate db");
+    let tx = db.begin_write().expect("retarget decision write");
+    let mut key = created::object_bytes(&c.object).expect("object key");
+    key.extend(c.prepared.version_id);
+    let mut version = Vec::new();
+    created::put_prepared(&mut version, &c.prepared);
+    version.extend(encode_id(id(1)));
+    version.extend(binding().revision.to_be_bytes());
+    tx.open_table(VERSIONS)
+        .expect("versions")
+        .insert(key.as_slice(), version.as_slice())
+        .expect("retarget version");
+    let mut event = created::object_bytes(&c.object).expect("event object");
+    created::put_prepared(&mut event, &c.prepared);
+    event.extend(encode_id(id(1)));
+    event.extend(created::binding_bytes(&binding()).expect("binding bytes"));
+    tx.open_table(EVENTS)
+        .expect("events")
+        .insert(c.operation_id.as_slice(), event.as_slice())
+        .expect("retarget event");
+    let canonical = c.encode().expect("canonical");
+    let mut operation = u64::try_from(canonical.len()).expect("length").to_be_bytes().to_vec();
+    operation.extend(canonical);
+    operation.extend(created::result_bytes(&success(&c, 1)));
+    tx.open_table(OPERATIONS)
+        .expect("operations")
+        .insert(c.operation_id.as_slice(), operation.as_slice())
+        .expect("retarget original response");
+    for index in 0..=1 {
+        let mut result = encode_id(id(index));
+        result.extend(created::result_bytes(&success(&c, 1)));
+        tx.open_table(APPLY_RESULTS)
+            .expect("results")
+            .insert(index, result.as_slice())
+            .expect("retarget duplicate response");
+    }
+    tx.commit().expect("coherent retarget commit");
+    drop(db);
+    assert!(
+        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
+        "U11 original decision cannot retarget duplicate log"
+    );
+}
+#[test]
+fn created_process_child() {
+    if let Some(path) = std::env::var_os("U11_CHILD_DB") {
+        let mut s = Store::initialize(path, identity()).expect("child init");
+        s.pin_capture_binding(binding()).expect("child pin");
+        assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
+        complete(&s);
+        println!("U11_ACK");
+        std::io::stdout().flush().expect("flush ACK");
+        loop {
+            std::thread::park()
+        }
+    }
+    let (_dir, mut s) = fixture();
+    assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
+    complete(&s);
+}
+#[test]
+fn created_acknowledged_apply_survives_child_kill() {
+    let dir = tempfile::tempdir().expect("dir");
+    let path = dir.path().join("db");
+    let mut child = Command::new(std::env::current_exe().expect("test executable"))
+        .args(["--exact", "application_tests::created_process_child", "--nocapture"])
+        .env("U11_CHILD_DB", &path)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit())
+        .spawn()
+        .expect("spawn");
+    let stdout = child.stdout.take().expect("stdout");
+    let (tx, rx) = mpsc::channel();
+    let reader = std::thread::spawn(move || {
+        for line in BufReader::new(stdout).lines() {
+            let line = line.expect("child line");
+            if line.ends_with("U11_ACK") {
+                let _ = tx.send(());
+                break;
+            }
+        }
+    });
+    let acknowledged = rx.recv_timeout(Duration::from_secs(20));
+    child.kill().expect("kill child");
+    child.wait().expect("wait child");
+    reader.join().expect("reader cleanup");
+    assert!(acknowledged.is_ok(), "U11 bounded child ACK missing");
+    complete(&reopen(&path));
+}
+#[test]
+fn created_power_cut_before_and_after_apply_flush() {
+    let (dir, mut s) = fixture();
+    append(&mut s, 0, &command());
+    s.save_committed(id(0)).expect("committed");
+    drop(s);
+    let path = dir.path().join("db");
+    let backend = CrashBackend::new(&path);
+    let mut s = Store::from_database(backend.database(), identity()).expect("backend");
+    assert_eq!(s.apply_next_created(id(0)).expect("apply ACK"), success(&command(), 0));
+    backend.cut();
+    drop(s);
+    complete(&reopen(&path));
+    for sync in 1..=2 {
+        for after in [false, true] {
+            let (dir, mut s) = fixture();
+            append(&mut s, 0, &command());
+            s.save_committed(id(0)).expect("committed");
+            drop(s);
+            let path = dir.path().join("db");
+            let backend = CrashBackend::new(&path);
+            let mut s = Store::from_database(backend.database(), identity()).expect("backend");
+            let (entered, resume) = backend.pause_sync(sync, after);
+            let writer = std::thread::spawn(move || {
+                let result = s.apply_next_created(id(0));
+                (s, result)
+            });
+            entered.wait();
+            backend.cut();
+            resume.wait();
+            let (s, result) = writer.join().expect("writer");
+            drop(s);
+            let reopened = Store::open(&path, identity());
+            assert!(
+                reopened.is_ok(),
+                "U11 apply cut must reopen whole old/new state sync={sync} after={after}: {:?}",
+                reopened.as_ref().err()
+            );
+            let s = reopened.expect("asserted reopen");
+            if s.last_applied().expect("last").is_some() {
+                complete(&s)
+            } else {
+                assert!(result.is_err());
+                assert_eq!(s.created_head(&command().object).expect("old head"), None);
+                assert_eq!(
+                    s.created_version(&command().object, command().prepared.version_id)
+                        .expect("old version"),
+                    None
+                );
+                assert_eq!(s.created_event(command().operation_id).expect("old event"), None);
+                assert_eq!(s.created_operation(command().operation_id).expect("old op"), None);
+                assert_eq!(s.created_apply_result(0).expect("old result"), None)
+            }
+            eprintln!("U11 cut boundary sync={sync} after={after}");
+        }
+    }
+}

crates/capture-authority/src/committed.rs

@@ -0,0 +1,63 @@
+use crate::{LOGS, LogId, META, Store, StoreError, decode_id, decode_log, encode_id, number};
+use redb::{ReadableDatabase, ReadableTable};
+
+pub(crate) fn validate_committed(
+    meta: &impl ReadableTable<&'static str, &'static [u8]>,
+    logs: &impl ReadableTable<u64, &'static [u8]>,
+) -> Result<Option<LogId>, StoreError> {
+    let schema = number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?;
+    let marker = meta.get("committed")?;
+    match (schema, marker) {
+        (1 | 3 | 4, None) => {
+            if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
+                return Err(StoreError::CorruptRecord);
+            }
+            Ok(None)
+        }
+        (2 | 3 | 4, Some(marker)) => {
+            let id = decode_id(marker.value())?;
+            let record = logs.get(id.index)?.ok_or(StoreError::CorruptRecord)?;
+            if decode_log(record.value())?.id != id {
+                return Err(StoreError::CorruptRecord);
+            }
+            Ok(Some(id))
+        }
+        (1 | 2, _) => Err(StoreError::CorruptRecord),
+        _ => Err(StoreError::UnsupportedSchema),
+    }
+}
+
+impl Store {
+    /// Read the exact committed boundary, validating it against the stored log.
+    pub fn read_committed(&self) -> Result<Option<LogId>, StoreError> {
+        let tx = self.database.begin_read()?;
+        validate_committed(&tx.open_table(META)?, &tx.open_table(LOGS)?)
+    }
+
+    /// Persist a monotonic committed boundary supplied by the authority owner.
+    pub fn save_committed(&mut self, id: LogId) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let logs = tx.open_table(LOGS)?;
+            let previous = validate_committed(&meta, &logs)?;
+            let record = logs.get(id.index)?.ok_or(StoreError::InvalidCommittedLog)?;
+            if decode_log(record.value())?.id != id {
+                return Err(StoreError::InvalidCommittedLog);
+            }
+            if let Some(previous) = previous {
+                if id.index < previous.index {
+                    return Err(StoreError::CommittedRegression);
+                }
+                if id == previous {
+                    return Ok(());
+                }
+            }
+            meta.insert("committed", encode_id(id).as_slice())?;
+            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3 | 4) {
+                meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
+            }
+        }
+        self.commit(tx)
+    }
+}

crates/capture-authority/src/created.rs

@@ -0,0 +1,242 @@
+use crate::{LogId, StoreError, decode_id, encode_id};
+
+pub const MAX_OBJECT_KEY_LENGTH: usize = 4096;
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ObjectIdentity {
+    pub bucket_incarnation: [u8; 16],
+    pub key: Vec<u8>,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PreparedIdentity {
+    pub version_id: [u8; 16],
+    pub preparation_id: [u8; 16],
+    pub content_digest: [u8; 32],
+    pub content_length: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DecideCreated {
+    pub operation_id: [u8; 16],
+    pub object: ObjectIdentity,
+    pub expected_head: Option<[u8; 16]>,
+    pub prepared: PreparedIdentity,
+    pub binding_revision: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CaptureBinding {
+    pub revision: u64,
+    pub binding_id: [u8; 16],
+    pub target_id: [u8; 16],
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CreatedResult {
+    Created {
+        version_id: [u8; 16],
+        event_id: [u8; 16],
+        decision_log_id: LogId,
+    },
+    HeadMismatch {
+        actual_head: Option<[u8; 16]>,
+    },
+    OperationConflict,
+    BindingMismatch,
+    VersionConflict,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedVersion {
+    pub prepared: PreparedIdentity,
+    pub decision_log_id: LogId,
+    pub binding_revision: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedEvent {
+    pub object: ObjectIdentity,
+    pub prepared: PreparedIdentity,
+    pub decision_log_id: LogId,
+    pub binding: CaptureBinding,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedOperation {
+    pub canonical_command: Vec<u8>,
+    pub result: CreatedResult,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedApplyResult {
+    pub log_id: LogId,
+    pub result: CreatedResult,
+}
+
+pub(crate) struct Decoder<'a> {
+    bytes: &'a [u8],
+}
+impl<'a> Decoder<'a> {
+    pub(crate) fn new(bytes: &'a [u8]) -> Self {
+        Self { bytes }
+    }
+    pub(crate) fn take<const N: usize>(&mut self) -> Result<[u8; N], StoreError> {
+        let a = self
+            .bytes
+            .get(..N)
+            .ok_or(StoreError::CorruptRecord)?
+            .try_into()
+            .map_err(|_| StoreError::CorruptRecord)?;
+        self.bytes = &self.bytes[N..];
+        Ok(a)
+    }
+    pub(crate) fn u64(&mut self) -> Result<u64, StoreError> {
+        Ok(u64::from_be_bytes(self.take()?))
+    }
+    pub(crate) fn uuid(&mut self) -> Result<[u8; 16], StoreError> {
+        let id = self.take()?;
+        if id == [0; 16] {
+            return Err(StoreError::CorruptRecord);
+        }
+        Ok(id)
+    }
+    pub(crate) fn option(&mut self) -> Result<Option<[u8; 16]>, StoreError> {
+        match self.take::<1>()?[0] {
+            0 => Ok(None),
+            1 => Ok(Some(self.uuid()?)),
+            _ => Err(StoreError::CorruptRecord),
+        }
+    }
+    pub(crate) fn object(&mut self) -> Result<ObjectIdentity, StoreError> {
+        let bucket_incarnation = self.uuid()?;
+        let len = usize::try_from(u32::from_be_bytes(self.take()?)).map_err(|_| StoreError::CorruptRecord)?;
+        if len > MAX_OBJECT_KEY_LENGTH {
+            return Err(StoreError::CorruptRecord);
+        }
+        let key = self.bytes.get(..len).ok_or(StoreError::CorruptRecord)?.to_vec();
+        self.bytes = &self.bytes[len..];
+        Ok(ObjectIdentity { bucket_incarnation, key })
+    }
+    pub(crate) fn prepared(&mut self) -> Result<PreparedIdentity, StoreError> {
+        Ok(PreparedIdentity {
+            version_id: self.uuid()?,
+            preparation_id: self.uuid()?,
+            content_digest: self.take()?,
+            content_length: self.u64()?,
+        })
+    }
+    pub(crate) fn id(&mut self) -> Result<LogId, StoreError> {
+        decode_id(&self.take::<24>()?)
+    }
+    pub(crate) fn binding(&mut self) -> Result<CaptureBinding, StoreError> {
+        Ok(CaptureBinding {
+            revision: self.u64()?,
+            binding_id: self.uuid()?,
+            target_id: self.uuid()?,
+        })
+    }
+    pub(crate) fn finish(self) -> Result<(), StoreError> {
+        if self.bytes.is_empty() {
+            Ok(())
+        } else {
+            Err(StoreError::CorruptRecord)
+        }
+    }
+    pub(crate) fn result(&mut self) -> Result<CreatedResult, StoreError> {
+        match self.take::<1>()?[0] {
+            0 => Ok(CreatedResult::Created {
+                version_id: self.uuid()?,
+                event_id: self.uuid()?,
+                decision_log_id: self.id()?,
+            }),
+            1 => Ok(CreatedResult::HeadMismatch {
+                actual_head: self.option()?,
+            }),
+            2 => Ok(CreatedResult::OperationConflict),
+            3 => Ok(CreatedResult::BindingMismatch),
+            4 => Ok(CreatedResult::VersionConflict),
+            _ => Err(StoreError::CorruptRecord),
+        }
+    }
+}
+pub(crate) fn put_option(b: &mut Vec<u8>, id: Option<[u8; 16]>) {
+    b.push(u8::from(id.is_some()));
+    if let Some(id) = id {
+        b.extend(id)
+    }
+}
+pub(crate) fn object_bytes(o: &ObjectIdentity) -> Result<Vec<u8>, StoreError> {
+    if o.bucket_incarnation == [0; 16] || o.key.len() > MAX_OBJECT_KEY_LENGTH {
+        return Err(StoreError::CorruptRecord);
+    }
+    let capacity = 20usize.checked_add(o.key.len()).ok_or(StoreError::CorruptRecord)?;
+    let mut b = Vec::with_capacity(capacity);
+    b.extend(o.bucket_incarnation);
+    b.extend(
+        u32::try_from(o.key.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes(),
+    );
+    b.extend(&o.key);
+    Ok(b)
+}
+pub(crate) fn put_prepared(b: &mut Vec<u8>, p: &PreparedIdentity) {
+    b.extend(p.version_id);
+    b.extend(p.preparation_id);
+    b.extend(p.content_digest);
+    b.extend(p.content_length.to_be_bytes())
+}
+pub(crate) fn binding_bytes(v: &CaptureBinding) -> Result<Vec<u8>, StoreError> {
+    let mut b = v.revision.to_be_bytes().to_vec();
+    b.extend(v.binding_id);
+    b.extend(v.target_id);
+    let mut d = Decoder::new(&b);
+    d.binding()?;
+    d.finish()?;
+    Ok(b)
+}
+pub(crate) fn result_bytes(r: &CreatedResult) -> Vec<u8> {
+    let mut b = Vec::new();
+    match r {
+        CreatedResult::Created {
+            version_id,
+            event_id,
+            decision_log_id,
+        } => {
+            b.push(0);
+            b.extend(version_id);
+            b.extend(event_id);
+            b.extend(encode_id(*decision_log_id))
+        }
+        CreatedResult::HeadMismatch { actual_head } => {
+            b.push(1);
+            put_option(&mut b, *actual_head)
+        }
+        CreatedResult::OperationConflict => b.push(2),
+        CreatedResult::BindingMismatch => b.push(3),
+        CreatedResult::VersionConflict => b.push(4),
+    }
+    b
+}
+impl DecideCreated {
+    pub fn encode(&self) -> Result<Vec<u8>, StoreError> {
+        let object = object_bytes(&self.object)?;
+        let capacity = object.len().checked_add(134).ok_or(StoreError::CorruptRecord)?;
+        let mut b = Vec::with_capacity(capacity);
+        b.extend(b"CRTD\x01");
+        b.extend(self.operation_id);
+        b.extend(object);
+        put_option(&mut b, self.expected_head);
+        put_prepared(&mut b, &self.prepared);
+        b.extend(self.binding_revision.to_be_bytes());
+        Self::decode(&b)?;
+        Ok(b)
+    }
+    pub fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
+        let mut d = Decoder::new(bytes);
+        if d.take::<5>()? != *b"CRTD\x01" {
+            return Err(StoreError::CorruptRecord);
+        }
+        let c = Self {
+            operation_id: d.uuid()?,
+            object: d.object()?,
+            expected_head: d.option()?,
+            prepared: d.prepared()?,
+            binding_revision: d.u64()?,
+        };
+        d.finish()?;
+        Ok(c)
+    }
+}

crates/capture-authority/src/node.rs

@@ -0,0 +1,371 @@
+//! Validated capture peer metadata and its independent CNOD v1 codec.
+use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct};
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct CaptureNode {
+    rpc_uri: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum CaptureNodeError {
+    #[error("invalid RPC URI")]
+    InvalidRpcUri,
+    #[error("invalid RPC URI")]
+    InvalidRpcUriParse(#[source] url::ParseError),
+    #[error("RPC URI too long")]
+    RpcUriTooLong,
+    #[error("invalid node magic")]
+    InvalidNodeMagic,
+    #[error("unsupported node version")]
+    UnsupportedNodeVersion,
+    #[error("invalid node length")]
+    InvalidNodeLength,
+    #[error("truncated node record")]
+    TruncatedNodeRecord,
+    #[error("invalid node UTF-8")]
+    InvalidNodeUtf8(#[source] std::str::Utf8Error),
+    #[error("trailing node bytes")]
+    TrailingNodeBytes,
+}
+impl CaptureNodeError {
+    pub fn code(&self) -> &'static str {
+        match self {
+            Self::InvalidRpcUri | Self::InvalidRpcUriParse(_) => "invalid_rpc_uri",
+            Self::RpcUriTooLong => "rpc_uri_too_long",
+            Self::InvalidNodeMagic => "invalid_node_magic",
+            Self::UnsupportedNodeVersion => "unsupported_node_version",
+            Self::InvalidNodeLength => "invalid_node_length",
+            Self::TruncatedNodeRecord => "truncated_node_record",
+            Self::InvalidNodeUtf8(_) => "invalid_node_utf8",
+            Self::TrailingNodeBytes => "trailing_node_bytes",
+        }
+    }
+}
+impl CaptureNode {
+    pub fn new(rpc_uri: String) -> Result<Self, CaptureNodeError> {
+        let node = Self { rpc_uri };
+        node.validate()?;
+        Ok(node)
+    }
+
+    pub fn rpc_uri(&self) -> &str {
+        &self.rpc_uri
+    }
+
+    pub fn validate(&self) -> Result<(), CaptureNodeError> {
+        let uri = &self.rpc_uri;
+        if uri.len() > 4096 {
+            return Err(CaptureNodeError::RpcUriTooLong);
+        }
+        if uri.is_empty() || uri.chars().any(|c| c.is_whitespace() || c.is_control() || c == '\\') {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let rest = uri
+            .strip_prefix("http://")
+            .or_else(|| uri.strip_prefix("https://"))
+            .ok_or(CaptureNodeError::InvalidRpcUri)?;
+        if uri.contains('?') || uri.contains('#') {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let (authority, path) = rest.split_once('/').map_or((rest, ""), |(a, p)| (a, p));
+        // An empty suffix after the first slash is the only allowed path.
+        if !path.is_empty() || authority.contains('@') {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let (host, port) = if authority.starts_with('[') {
+            let end = authority.find(']').ok_or(CaptureNodeError::InvalidRpcUri)?;
+            let port = authority
+                .get(end + 1..)
+                .and_then(|suffix| suffix.strip_prefix(':'))
+                .ok_or(CaptureNodeError::InvalidRpcUri)?;
+            (&authority[..=end], port)
+        } else {
+            authority.split_once(':').ok_or(CaptureNodeError::InvalidRpcUri)?
+        };
+        if host.is_empty() || port.is_empty() || !port.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let port = port.parse::<u16>().map_err(|_| CaptureNodeError::InvalidRpcUri)?;
+        if port == 0 {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let parsed_url = url::Url::parse(uri).map_err(CaptureNodeError::InvalidRpcUriParse)?;
+        if parsed_url.host().is_none() {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        Ok(())
+    }
+
+    pub fn encode(&self) -> Result<Vec<u8>, CaptureNodeError> {
+        self.validate()?;
+        let length = u32::try_from(self.rpc_uri.len()).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
+        let mut bytes = b"CNOD\x01".to_vec();
+        bytes.extend_from_slice(&length.to_be_bytes());
+        bytes.extend_from_slice(self.rpc_uri.as_bytes());
+        Ok(bytes)
+    }
+
+    pub fn decode(bytes: &[u8]) -> Result<Self, CaptureNodeError> {
+        if bytes.len() < 9 {
+            return Err(CaptureNodeError::TruncatedNodeRecord);
+        }
+        if &bytes[..4] != b"CNOD" {
+            return Err(CaptureNodeError::InvalidNodeMagic);
+        }
+        if bytes[4] != 1 {
+            return Err(CaptureNodeError::UnsupportedNodeVersion);
+        }
+        let length = u32::from_be_bytes(bytes[5..9].try_into().map_err(|_| CaptureNodeError::InvalidNodeLength)?);
+        if !(1..=4096).contains(&length) {
+            return Err(CaptureNodeError::InvalidNodeLength);
+        }
+        let length = usize::try_from(length).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
+        let end = 9_usize.checked_add(length).ok_or(CaptureNodeError::InvalidNodeLength)?;
+        if bytes.len() < end {
+            return Err(CaptureNodeError::TruncatedNodeRecord);
+        }
+        if bytes.len() != end {
+            return Err(CaptureNodeError::TrailingNodeBytes);
+        }
+        let uri = std::str::from_utf8(&bytes[9..end]).map_err(CaptureNodeError::InvalidNodeUtf8)?;
+        Self::new(uri.to_owned())
+    }
+}
+impl Serialize for CaptureNode {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        self.validate().map_err(serde::ser::Error::custom)?;
+        let mut state = serializer.serialize_struct("CaptureNode", 1)?;
+        state.serialize_field("rpc_uri", &self.rpc_uri)?;
+        state.end()
+    }
+}
+impl<'de> Deserialize<'de> for CaptureNode {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        #[derive(Deserialize)]
+        #[serde(deny_unknown_fields)]
+        struct Fields {
+            rpc_uri: String,
+        }
+        let fields = Fields::deserialize(deserializer).map_err(|_| serde::de::Error::custom("invalid capture node fields"))?;
+        Self::new(fields.rpc_uri).map_err(serde::de::Error::custom)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::error::Error;
+
+    fn record(payload: &[u8]) -> Vec<u8> {
+        let mut bytes = b"CNOD\x01".to_vec();
+        bytes.extend_from_slice(&u32::try_from(payload.len()).expect("test payload length").to_be_bytes());
+        bytes.extend_from_slice(payload);
+        bytes
+    }
+    fn reject_uri(uri: &str) {
+        let result = CaptureNode::new(uri.to_owned());
+        assert!(result.is_err(), "URI should reject: {uri:?}");
+        assert_eq!(result.expect_err("rejected URI").code(), "invalid_rpc_uri");
+    }
+    fn reject_record(bytes: &[u8], code: &str) {
+        let result = CaptureNode::decode(bytes);
+        assert!(result.is_err(), "record should reject with {code}");
+        assert_eq!(result.expect_err("rejected record").code(), code);
+    }
+    #[test]
+    fn actual_openraft_node_bound() {
+        fn require_node<T: openraft::Node>() {}
+        require_node::<CaptureNode>();
+        assert_eq!(CaptureNode::default().rpc_uri(), "");
+    }
+    #[test]
+    fn default_and_empty_rejected_at_every_boundary() {
+        let node = CaptureNode::default();
+        assert_eq!(node.rpc_uri(), "");
+        reject_uri("");
+        assert_eq!(node.validate().expect_err("empty validation").code(), "invalid_rpc_uri");
+        assert!(serde_json::to_string(&node).is_err());
+        assert!(serde_json::from_str::<CaptureNode>(r#"{"rpc_uri":""}"#).is_err());
+        assert_eq!(node.encode().expect_err("empty encoding").code(), "invalid_rpc_uri");
+        reject_record(&record(b""), "invalid_node_length");
+    }
+    #[test]
+    fn accepted_uris_preserve_exact_bytes() {
+        for uri in [
+            "http://node:80",
+            "https://node:443",
+            "http://node:1",
+            "https://node:65535/",
+            "http://127.0.0.1:9000",
+            "https://[::1]:443/",
+            "http://[2001:db8::1]:1",
+            "http://NoDe:00080/",
+        ] {
+            let result = CaptureNode::new(uri.to_owned());
+            assert!(result.is_ok(), "valid URI construction: {uri}");
+            let node = result.expect("valid URI");
+            assert_eq!(node.rpc_uri(), uri, "preserve explicit default port and raw URI");
+            assert_eq!(node.encode().expect("encode URI"), record(uri.as_bytes()));
+            assert_eq!(CaptureNode::decode(&record(uri.as_bytes())).expect("decode URI"), node);
+        }
+    }
+    #[test]
+    fn golden_cnod_bytes() {
+        let node = CaptureNode {
+            rpc_uri: "http://node:80".to_owned(),
+        };
+        let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
+        assert_eq!(node.encode().ok().as_deref(), Some(golden.as_slice()), "golden CNOD bytes");
+    }
+    #[test]
+    fn valid_cnod_decode() {
+        let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
+        let decoded = CaptureNode::decode(golden);
+        assert!(decoded.is_ok(), "golden decode succeeds");
+        assert_eq!(decoded.expect("golden decoded node").rpc_uri(), "http://node:80");
+    }
+    #[test]
+    fn serde_shape_roundtrip_and_strict_fields() {
+        let node = CaptureNode::new("https://node:443/".to_owned()).expect("serde node");
+        let json = r#"{"rpc_uri":"https://node:443/"}"#;
+        assert_eq!(serde_json::to_string(&node).expect("serialize node"), json);
+        assert_eq!(serde_json::from_str::<CaptureNode>(json).expect("deserialize node"), node);
+        for json in [
+            r#"{}"#,
+            r#"{"rpc_uri":1}"#,
+            r#"{"rpc_uri":null}"#,
+            r#"{"rpc_uri":"http://node:1","extra":0}"#,
+            r#"{"rpc_uri":"http://node:1","rpc_uri":"http://node:2"}"#,
+        ] {
+            assert!(serde_json::from_str::<CaptureNode>(json).is_err(), "strict fields: {json}");
+        }
+    }
+    #[test]
+    fn credentials_and_raw_uri_shapes_reject() {
+        for uri in [
+            "http://user@node:1",
+            "http://user:password@node:1",
+            "http://@node:1",
+            "HTTP://node:1",
+            "Https://node:1",
+            "http://node",
+            "http://node:",
+            "http://node:0",
+            "http://node:-1",
+            "http://node:+1",
+            "http://node:65536",
+            "http://node:999999999999999999999",
+            "http://[::1:1",
+            "http://[::1]extra:1",
+            "http://[invalid]:1",
+            "http://::1:1",
+            "http://:1",
+            "/disk/path",
+            "file:///disk/path",
+            "http://node:1/./",
+            "http://node:1/../",
+            "http://node:1/%2e/",
+            "http://node:1/%2E%2e/",
+            "http://node:1/path",
+            "http://node:1\\",
+            " http://node:1",
+            "http://node:1\n",
+            "http://no\tde:1",
+            "http://node:1\u{00a0}",
+            "http://node:1\u{0085}",
+            "http://node:1\0",
+        ] {
+            reject_uri(uri);
+        }
+    }
+    #[test]
+    fn query_and_fragment_markers_reject() {
+        for uri in [
+            "http://node:1?",
+            "http://node:1?x=y",
+            "http://node:1/?",
+            "http://node:1#",
+            "http://node:1#fragment",
+            "http://node:1/#",
+        ] {
+            reject_uri(uri);
+        }
+    }
+    #[test]
+    fn utf8_byte_length_boundaries() {
+        for length in [4095, 4096] {
+            let prefix = "http://node:";
+            let uri = format!("{prefix}{}1", "0".repeat(length - prefix.len() - 1));
+            assert_eq!(uri.len(), length);
+            assert!(url::Url::parse(&uri).is_ok(), "boundary fixture accepted by actual URL parser");
+            let node = CaptureNode::new(uri.clone()).expect("boundary node");
+            assert_eq!(node.rpc_uri(), uri);
+            assert_eq!(
+                CaptureNode::decode(&node.encode().expect("boundary encoding")).expect("boundary decoding"),
+                node
+            );
+        }
+        let uri = format!("http://node:{}1", "0".repeat(4097 - "http://node:".len() - 1));
+        assert_eq!(CaptureNode::new(uri).expect_err("oversize URI").code(), "rpc_uri_too_long");
+        let unicode = format!("http://{}:1", "é".repeat(2044));
+        assert!(unicode.chars().count() < 4096);
+        assert!(unicode.len() > 4096);
+        assert_eq!(CaptureNode::new(unicode).expect_err("UTF-8 byte limit").code(), "rpc_uri_too_long");
+    }
+    #[test]
+    fn cnod_headers_and_lengths_reject_exact_codes() {
+        let valid = record(b"http://node:80");
+        for end in 0..9 {
+            reject_record(&valid[..end], "truncated_node_record");
+        }
+        let mut bad = valid.clone();
+        bad[0] = b'X';
+        reject_record(&bad, "invalid_node_magic");
+        bad = valid.clone();
+        bad[4] = 2;
+        reject_record(&bad, "unsupported_node_version");
+        for length in [0_u32, 4097, u32::MAX] {
+            bad = b"CNOD\x01".to_vec();
+            bad.extend_from_slice(&length.to_be_bytes());
+            reject_record(&bad, "invalid_node_length");
+        }
+        for end in 9..valid.len() {
+            reject_record(&valid[..end], "truncated_node_record");
+        }
+        reject_record(&record(&[0xff]), "invalid_node_utf8");
+        reject_record(&record(&[0xc3, 0x28]), "invalid_node_utf8");
+        reject_record(&record(b"invalid URI"), "invalid_rpc_uri");
+    }
+    #[test]
+    fn trailing_bytes_reject() {
+        for tail in [&b"x"[..], &b"xyz"[..]] {
+            let mut bytes = record(b"http://node:80");
+            bytes.extend_from_slice(tail);
+            reject_record(&bytes, "trailing_node_bytes");
+        }
+    }
+    #[test]
+    fn errors_and_sources_redact_input() {
+        let sentinel = "http://SECRET_USER:SECRET_PASSWORD@node:1";
+        let error = CaptureNode::new(sentinel.to_owned()).expect_err("credentials reject");
+        let mut current: &dyn Error = &error;
+        loop {
+            let rendered = format!("{current} {current:?}");
+            assert!(!rendered.contains("SECRET"));
+            match current.source() {
+                Some(source) => current = source,
+                None => break,
+            }
+        }
+        let json = format!(r#"{{"rpc_uri":"{sentinel}"}}"#);
+        let error = serde_json::from_str::<CaptureNode>(&json).expect_err("serde credentials reject");
+        assert!(!format!("{error} {error:?}").contains("SECRET"));
+        let json = format!(r#"{{"{sentinel}":0}}"#);
+        let error = serde_json::from_str::<CaptureNode>(&json).expect_err("unknown fields reject");
+        assert!(!format!("{error} {error:?}").contains("SECRET"));
+        let error = CaptureNode::new("http://[bad]:1".to_owned()).expect_err("parser rejection");
+        assert!(error.source().is_some(), "retain URL parser source");
+        let error = CaptureNode::decode(&record(&[0xff])).expect_err("UTF-8 rejection");
+        assert!(error.source().is_some(), "retain UTF-8 source");
+    }
+}

crates/capture-authority/src/truncate.rs

@@ -0,0 +1,43 @@
+use crate::{LOGS, META, Store, StoreError, decode_id, decode_log, encode_id};
+use redb::ReadableTable;
+
+impl Store {
+    /// Atomically remove every log entry at or after `from`.
+    pub fn truncate_suffix(&mut self, from: u64) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let logs = tx.open_table(LOGS)?;
+            if crate::committed::validate_committed(&meta, &logs)?.is_some_and(|id| from <= id.index) {
+                return Err(StoreError::CommittedLogProtected);
+            }
+            drop(logs);
+            let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?;
+            if tail.is_none_or(|id| from > id.index) {
+                return Ok(());
+            }
+            let mut logs = tx.open_table(LOGS)?;
+            let new_tail = if from == 0 {
+                None
+            } else {
+                let index = from - 1;
+                let record = logs.get(index)?.ok_or(StoreError::InvalidLogSequence)?;
+                let id = decode_log(record.value())?.id;
+                if id.index != index {
+                    return Err(StoreError::CorruptRecord);
+                }
+                Some(id)
+            };
+            logs.retain(|index, _| index < from)?;
+            match new_tail {
+                Some(id) => {
+                    meta.insert("tail", encode_id(id).as_slice())?;
+                }
+                None => {
+                    meta.remove("tail")?;
+                }
+            }
+        }
+        self.commit(tx)
+    }
+}

crates/ecstore/src/api/mod.rs

@@ -371,6 +371,24 @@ pub mod config {
             with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock,
             with_server_config_write_lock,
         };
+
+        /// Read config bytes and metadata without treating an empty payload as absent.
+        pub async fn read_config_preserve_empty_with_metadata_opts(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            opts: &crate::object_api::ObjectOptions,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_preserve_empty_with_metadata_opts(api, file, opts).await
+        }
+
+        pub async fn read_config_limited_preserve_empty_with_metadata_opts(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            opts: &crate::object_api::ObjectOptions,
+            max_bytes: usize,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_limited_preserve_empty_with_metadata_opts(api, file, opts, max_bytes).await
+        }
     }
 
     pub mod storageclass {

crates/ecstore/src/config/com.rs

@@ -495,10 +495,22 @@ pub(crate) async fn read_config_preserve_empty<S>(api: Arc<S>, file: &str) -> Re
 where
     S: EcstoreObjectIO,
 {
-    let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, None).await?;
+    let (data, _obj) = read_config_preserve_empty_with_metadata_opts(api, file, &ObjectOptions::default()).await?;
     Ok(data)
 }
 
+/// Preserve an empty config payload, its metadata, and caller-supplied read options.
+pub(crate) async fn read_config_preserve_empty_with_metadata_opts<S>(
+    api: Arc<S>,
+    file: &str,
+    opts: &ObjectOptions,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    read_config_with_metadata_inner(api, file, opts, true, None).await
+}
+
 pub async fn read_config_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
 where
     S: ObjectIO<
@@ -527,15 +539,13 @@ pub(crate) async fn read_config_no_lock_preserve_empty_with_metadata<S>(api: Arc
 where
     S: EcstoreObjectIO,
 {
-    read_config_with_metadata_inner(
+    read_config_preserve_empty_with_metadata_opts(
         api,
         file,
         &ObjectOptions {
             no_lock: true,
             ..Default::default()
         },
-        true,
-        None,
     )
     .await
 }
@@ -2815,9 +2825,10 @@ mod tests {
         config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob,
         heal_config_descriptor, is_standard_object_server_config, lookup_configs, map_system_metadata_write_error,
         new_and_save_server_config, read_config, read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty,
-        read_config_with_metadata, read_config_without_migrate, read_server_config_snapshot, save_config_with_opts_inner,
-        save_server_config, save_server_config_snapshot, save_server_config_snapshot_with_generation,
-        server_config_transaction_lock_path, should_warn_ignored_scalar_section, storage_class_kvs_mut,
+        read_config_preserve_empty_with_metadata_opts, read_config_with_metadata, read_config_without_migrate,
+        read_server_config_snapshot, save_config_with_opts_inner, save_server_config, save_server_config_snapshot,
+        save_server_config_snapshot_with_generation, server_config_transaction_lock_path, should_warn_ignored_scalar_section,
+        storage_class_kvs_mut,
     };
     use crate::config::{audit, heal, notify, oidc, scanner};
     use crate::disk::{RUSTFS_META_BUCKET, endpoint::Endpoint};
@@ -5065,6 +5076,7 @@ mod tests {
         heal_replacement: Option<Vec<u8>>,
         heal_calls: AtomicUsize,
         write_calls: AtomicUsize,
+        last_get_no_lock: AtomicBool,
         last_put_no_lock: AtomicBool,
         last_put_preconditions: Mutex<Option<HTTPPreconditions>>,
         revision: AtomicUsize,
@@ -5080,6 +5092,7 @@ mod tests {
                 heal_replacement,
                 heal_calls: AtomicUsize::new(0),
                 write_calls: AtomicUsize::new(0),
+                last_get_no_lock: AtomicBool::new(false),
                 last_put_no_lock: AtomicBool::new(false),
                 last_put_preconditions: Mutex::new(None),
                 revision: AtomicUsize::new(1),
@@ -5135,8 +5148,9 @@ mod tests {
             _object: &str,
             _range: Option<HTTPRangeSpec>,
             _h: HeaderMap,
-            _opts: &ObjectOptions,
+            opts: &ObjectOptions,
         ) -> Result<GetObjectReader> {
+            self.last_get_no_lock.store(opts.no_lock, Ordering::SeqCst);
             let data = match &*self.state.lock().expect("state lock poisoned") {
                 RecoveryReadState::Missing => return Err(Error::ConfigNotFound),
                 RecoveryReadState::Blob(data) => data.clone(),
@@ -5431,6 +5445,60 @@ mod tests {
         assert!(data.is_empty());
     }
 
+    #[tokio::test]
+    async fn read_config_preserve_empty_metadata_keeps_empty_and_options() {
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));
+        for no_lock in [false, true] {
+            let opts = ObjectOptions {
+                no_lock,
+                ..Default::default()
+            };
+            let (data, info) = read_config_preserve_empty_with_metadata_opts(store.clone(), "config/empty.json", &opts)
+                .await
+                .expect("an existing empty config must retain its payload and metadata");
+            assert!(data.is_empty());
+            assert_eq!(info.size, 0);
+            assert_eq!(info.actual_size, 0);
+            assert_eq!(info.etag.as_deref(), Some("config-1"));
+            assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
+            assert_eq!(store.last_get_no_lock.load(Ordering::SeqCst), no_lock);
+        }
+        let err = read_config(store, "config/empty.json")
+            .await
+            .expect_err("the existing reader must still treat empty configs as missing");
+        assert!(matches!(err, Error::ConfigNotFound));
+    }
+
+    #[tokio::test]
+    async fn read_config_preserve_empty_metadata_keeps_payload_and_metadata() {
+        let payload = b"original config bytes".to_vec();
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(payload.clone()), None));
+        let (data, info) =
+            read_config_preserve_empty_with_metadata_opts(store, "config/nonempty.json", &ObjectOptions::default())
+                .await
+                .expect("a nonempty config must retain its payload and metadata");
+        assert_eq!(data, payload);
+        let size = i64::try_from(payload.len()).expect("test payload size fits i64");
+        assert_eq!(info.size, size);
+        assert_eq!(info.actual_size, size);
+        assert_eq!(info.etag.as_deref(), Some("config-1"));
+        assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
+    }
+
+    #[tokio::test]
+    async fn read_config_preserve_empty_metadata_preserves_read_errors() {
+        let missing = Arc::new(RecoveryMockStore::new(RecoveryReadState::Missing, None));
+        let err = read_config_preserve_empty_with_metadata_opts(missing, "config/missing.json", &ObjectOptions::default())
+            .await
+            .expect_err("physical absence must remain ConfigNotFound");
+        assert!(matches!(err, Error::ConfigNotFound));
+        let quorum = Arc::new(RecoveryMockStore::new(RecoveryReadState::QuorumError, None));
+        let err = read_config_preserve_empty_with_metadata_opts(quorum, "config/quorum.json", &ObjectOptions::default())
+            .await
+            .expect_err("a quorum failure must retain its typed read error");
+        assert!(matches!(err, Error::ErasureReadQuorum));
+    }
+
     #[async_trait::async_trait]
     impl StorageAdminApi for RecoveryMockStore {
         type BackendInfo = rustfs_madmin::BackendInfo;

crates/ecstore/src/disk/local.rs

@@ -12477,6 +12477,227 @@ mod test {
         }
     }
 
+    // SIGKILL tests process death on a healthy filesystem, not power loss.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn rename_data_process_kill_preserves_publication_boundary() {
+        use std::io::Write as _;
+        use std::os::unix::process::ExitStatusExt;
+        use std::process::{Child, Command, ExitStatus, Stdio};
+        use std::time::{Duration, Instant};
+
+        const TEST: &str = "disk::local::test::rename_data_process_kill_preserves_publication_boundary";
+        const ROLE: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROLE";
+        const ROOT: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROOT";
+        const NONCE: &str = "RUSTFS_TEST_PUBLICATION_KILL_NONCE";
+        const BUCKET: &str = "bucket";
+        const OBJECT: &str = "process-kill-object";
+        const STAGED: &str = "process-kill-staged";
+        const VERSION: Uuid = Uuid::from_u128(0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa);
+        const OLD: Uuid = Uuid::from_u128(0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb);
+        const NEW: Uuid = Uuid::from_u128(0xcccccccccccccccccccccccccccccccc);
+
+        struct ChildGuard(Child);
+        impl Drop for ChildGuard {
+            fn drop(&mut self) {
+                let _ = self.0.kill();
+                let _ = self.0.wait();
+            }
+        }
+
+        async fn wait_exit(child: &mut ChildGuard) -> ExitStatus {
+            let deadline = Instant::now() + Duration::from_secs(60);
+            loop {
+                if child.0.try_wait().expect("poll owned child").is_some() {
+                    return child.0.wait().expect("reap owned child");
+                }
+                assert!(Instant::now() < deadline, "owned child exceeded exit deadline");
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        }
+
+        fn spawn(root: &std::path::Path, nonce: &str, role: &str) -> ChildGuard {
+            let stdout = std::fs::File::create(root.join(format!("{role}.stdout"))).expect("child stdout file");
+            let stderr = std::fs::File::create(root.join(format!("{role}.stderr"))).expect("child stderr file");
+            ChildGuard(
+                Command::new(std::env::current_exe().expect("test executable"))
+                    .args(["--exact", TEST, "--nocapture"])
+                    .env(ROLE, role)
+                    .env(ROOT, root)
+                    .env(NONCE, nonce)
+                    .stdout(Stdio::from(stdout))
+                    .stderr(Stdio::from(stderr))
+                    .spawn()
+                    .expect("spawn owned publication child"),
+            )
+        }
+
+        fn child_logs(root: &std::path::Path, role: &str) -> String {
+            let stdout = std::fs::read_to_string(root.join(format!("{role}.stdout"))).expect("child stdout");
+            let stderr = std::fs::read_to_string(root.join(format!("{role}.stderr"))).expect("child stderr");
+            format!("{stdout} {stderr}")
+        }
+
+        fn file_info(data_dir: Uuid) -> FileInfo {
+            let mut fi = test_file_info(OBJECT, VERSION, Some(data_dir), None);
+            fi.size = 8;
+            fi.parts[0].size = 8;
+            fi.parts[0].actual_size = 8;
+            fi
+        }
+
+        if let Some(role) = std::env::var_os(ROLE) {
+            let role = role.to_str().expect("child role is UTF-8");
+            assert!(matches!(role, "writer-before" | "writer-after" | "verify-before" | "verify-after"));
+            let root = std::path::PathBuf::from(std::env::var_os(ROOT).expect("private child root"));
+            let nonce = std::env::var(NONCE).expect("private child nonce");
+            Uuid::parse_str(&nonce).expect("nonce is a UUID");
+            assert_eq!(root.file_name().and_then(|name| name.to_str()), Some(nonce.as_str()));
+            assert_eq!(std::fs::read_to_string(root.join("control-nonce")).expect("bound root nonce"), nonce);
+            let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
+            let _strict = durability_mode_override::set(DurabilityMode::Strict);
+            let disk = LocalDisk::new(&endpoint, false).await.expect("child opens real disk");
+            let object_dir = root.join(BUCKET).join(OBJECT);
+            if role.starts_with("verify-") {
+                let fi = disk
+                    .read_version("", BUCKET, OBJECT, &VERSION.to_string(), &ReadOptions::default())
+                    .await
+                    .expect("fresh verifier reads actual version");
+                let (expected_dir, expected_bytes) = if role == "verify-before" {
+                    assert_eq!(
+                        fs::read(object_dir.join(STORAGE_FORMAT_FILE))
+                            .await
+                            .expect("read old metadata"),
+                        fs::read(root.join("old-meta")).await.expect("read old fixture"),
+                        "prepublication metadata must remain byte-for-byte old"
+                    );
+                    (OLD, b"old-data")
+                } else {
+                    (NEW, b"new-data")
+                };
+                assert_eq!(fi.data_dir, Some(expected_dir), "{role}: actual reopened version data_dir");
+                let part = object_dir
+                    .join(fi.data_dir.expect("referenced data dir").to_string())
+                    .join("part.1");
+                assert_eq!(fs::read(part).await.expect("read referenced shard"), expected_bytes);
+                return;
+            }
+
+            let staged = root.join(RUSTFS_META_TMP_BUCKET).join(STAGED).join(NEW.to_string());
+            fs::create_dir_all(&staged).await.expect("staged shard directory");
+            fs::write(staged.join("part.1"), b"new-data")
+                .await
+                .expect("stage new shard bytes");
+            let ready = root.join(format!("{role}.ready"));
+            let pause = move || {
+                let pending = ready.with_extension("pending");
+                std::fs::write(&pending, &nonce).expect("write actual writer cut marker");
+                std::fs::rename(pending, &ready).expect("announce complete writer cut marker");
+                let (_sender, receiver) = std::sync::mpsc::channel::<()>();
+                receiver.recv().expect("writer remains paused until SIGKILL");
+            };
+            let _hook = if role == "writer-before" {
+                // Use the executor's exact path spelling, including canonical
+                // macOS roots and Linux mount-pinned I/O roots.
+                let destination = disk
+                    .io_get_object_path(BUCKET, &format!("{OBJECT}/{STORAGE_FORMAT_FILE}"))
+                    .expect("actual metadata publication path");
+                assert_eq!(
+                    std::fs::canonicalize(&destination).expect("canonical publication destination"),
+                    disk.root.join(BUCKET).join(OBJECT).join(STORAGE_FORMAT_FILE),
+                    "hook destination must identify this disk's seeded metadata"
+                );
+                Some(os::prepared_publication_test_hooks::install_at(
+                    os::prepared_publication_test_hooks::Stage::PreparedRename,
+                    &destination,
+                    pause,
+                ))
+            } else {
+                disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
+                    .await
+                    .expect("Strict real rename_data completes before cut");
+                pause();
+                return;
+            };
+            disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
+                .await
+                .expect("before writer must remain in physical hook");
+            panic!("prepublication hook unexpectedly returned");
+        }
+
+        let temp = tempfile::tempdir().expect("parent owns private disk roots");
+        for phase in ["before", "after"] {
+            let nonce = Uuid::new_v4().to_string();
+            let root = temp.path().join(&nonce);
+            fs::create_dir(&root).await.expect("private scenario root");
+            fs::write(root.join("control-nonce"), &nonce)
+                .await
+                .expect("bind private root");
+            let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
+            let disk = LocalDisk::new(&endpoint, false).await.expect("seed real disk");
+            ensure_test_volume(&disk, BUCKET).await;
+            ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+            let object_dir = root.join(BUCKET).join(OBJECT);
+            fs::create_dir_all(object_dir.join(OLD.to_string()))
+                .await
+                .expect("old shard directory");
+            fs::write(object_dir.join(OLD.to_string()).join("part.1"), b"old-data")
+                .await
+                .expect("seed old shard bytes");
+            let old_meta = test_meta(file_info(OLD));
+            fs::write(object_dir.join(STORAGE_FORMAT_FILE), &old_meta)
+                .await
+                .expect("seed old metadata");
+            fs::write(root.join("old-meta"), &old_meta)
+                .await
+                .expect("retain exact old fixture");
+            drop(disk);
+
+            let writer_role = format!("writer-{phase}");
+            let mut writer = spawn(&root, &nonce, &writer_role);
+            let deadline = Instant::now() + Duration::from_secs(60);
+            let ready = root.join(format!("{writer_role}.ready"));
+            loop {
+                if let Some(status) = writer.0.try_wait().expect("poll writer") {
+                    panic!(
+                        "writer exited before cut: {writer_role} status={status}: {}",
+                        child_logs(&root, &writer_role)
+                    );
+                }
+                match fs::read_to_string(&ready).await {
+                    Ok(marker) => {
+                        assert_eq!(marker, nonce, "cut marker belongs to this private writer");
+                        break;
+                    }
+                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+                    Err(error) => panic!("read cut marker: {error}"),
+                }
+                if Instant::now() >= deadline {
+                    panic!("writer cut deadline: {writer_role}: {}", child_logs(&root, &writer_role));
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+            let pid = writer.0.id();
+            writer.0.kill().expect("SIGKILL owned writer at verified cut");
+            let status = wait_exit(&mut writer).await;
+            assert_eq!(status.signal(), Some(9), "owned writer must actually die by SIGKILL");
+            writeln!(std::io::stdout(), "{writer_role} pid={pid} signal={:?}", status.signal())
+                .expect("record actual writer wait status");
+
+            let verifier_role = format!("verify-{phase}");
+            let mut verifier = spawn(&root, &nonce, &verifier_role);
+            let status = wait_exit(&mut verifier).await;
+            let logs = child_logs(&root, &verifier_role);
+            assert!(status.success(), "{verifier_role} status={status}: {logs}");
+            let stdout = std::fs::read_to_string(root.join(format!("{verifier_role}.stdout"))).expect("verifier stdout");
+            assert!(
+                stdout.contains(TEST) && stdout.contains("1 passed"),
+                "verifier must execute exact test: {logs}"
+            );
+            writeln!(std::io::stdout(), "{verifier_role} status={status}").expect("record actual verifier exit");
+        }
+    }
+
     /// Crash-consistency harness for the rename_data commit sequence
     /// (rustfs/backlog#935 HP-14, test plan rustfs/backlog#896; hard rule from
     /// rustfs/backlog#878: "After partial commit, the object can only be an old or new version; it cannot be mixed").

crates/iam/src/manager.rs

@@ -3345,27 +3345,23 @@ mod tests {
 
     #[test]
     fn test_extract_jwt_claims_basic() {
-        let user_identity = UserIdentity {
-            version: 1,
-            credentials: Credentials {
-                access_key: "test-access-key".to_string(),
-                secret_key: "test-secret-key".to_string(),
-                session_token: "invalid-token".to_string(), // Invalid token for testing error handling
-                expiration: None,
-                status: "enabled".to_string(),
-                parent_user: "".to_string(),
-                groups: None,
-                claims: Some({
-                    let mut claims = HashMap::new();
-                    claims.insert("sub".to_string(), json!("test-user"));
-                    claims.insert("aud".to_string(), json!("test-audience"));
-                    claims
-                }),
-                name: None,
-                description: None,
-            },
-            update_at: Some(OffsetDateTime::now_utc()),
-        };
+        let user_identity = UserIdentity::new(Credentials {
+            access_key: "test-access-key".to_string(),
+            secret_key: "test-secret-key".to_string(),
+            session_token: "invalid-token".to_string(), // Invalid token for testing error handling
+            expiration: None,
+            status: "enabled".to_string(),
+            parent_user: "".to_string(),
+            groups: None,
+            claims: Some({
+                let mut claims = HashMap::new();
+                claims.insert("sub".to_string(), json!("test-user"));
+                claims.insert("aud".to_string(), json!("test-audience"));
+                claims
+            }),
+            name: None,
+            description: None,
+        });
 
         let result = extract_jwt_claims(&user_identity);
         // In test environment without proper JWT setup, this should fail
@@ -3374,22 +3370,18 @@ mod tests {
 
     #[test]
     fn test_extract_jwt_claims_no_claims() {
-        let user_identity = UserIdentity {
-            version: 1,
-            credentials: Credentials {
-                access_key: "test-access-key".to_string(),
-                secret_key: "test-secret-key".to_string(),
-                session_token: "".to_string(), // Empty token
-                expiration: None,
-                status: "enabled".to_string(),
-                parent_user: "".to_string(),
-                groups: None,
-                claims: None,
-                name: None,
-                description: None,
-            },
-            update_at: Some(OffsetDateTime::now_utc()),
-        };
+        let user_identity = UserIdentity::new(Credentials {
+            access_key: "test-access-key".to_string(),
+            secret_key: "test-secret-key".to_string(),
+            session_token: "".to_string(), // Empty token
+            expiration: None,
+            status: "enabled".to_string(),
+            parent_user: "".to_string(),
+            groups: None,
+            claims: None,
+            name: None,
+            description: None,
+        });
 
         let result = extract_jwt_claims(&user_identity);
         // Should fail with empty session token
@@ -3448,11 +3440,7 @@ mod tests {
             description: None,
         };
 
-        let user_identity = UserIdentity {
-            version: 1,
-            credentials,
-            update_at: Some(OffsetDateTime::now_utc()),
-        };
+        let user_identity = UserIdentity::new(credentials);
 
         // Test basic structure
         assert_eq!(user_identity.version, 1);

crates/iam/src/storage_api.rs

@@ -91,6 +91,7 @@ pub(crate) mod crate_boundary {
 
 pub(crate) mod object_store {
     pub(crate) use super::storage_contracts::{HTTPPreconditions, ListOperations, ObjectInfoOrErr, ObjectOperations};
+    pub(crate) use rustfs_ecstore::api::config::com::read_config_preserve_empty_with_metadata_opts;
 }
 
 pub(crate) mod runtime {

crates/iam/src/store/object.rs

@@ -42,6 +42,7 @@ use tracing::{debug, error, warn};
 
 use crate::storage_api::object_store::{
     HTTPPreconditions, ListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, ObjectOperations,
+    read_config_preserve_empty_with_metadata_opts,
 };
 
 pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam"));
@@ -414,9 +415,14 @@ impl ObjectStore {
         &self,
         path: impl AsRef<str> + Send,
         mode: LoadMode,
+        preserve_empty: bool,
     ) -> Result<(Vec<u8>, IamObjectInfo)> {
         let path_ref = path.as_ref();
-        let (data, obj) = read_iam_config_with_metadata(self.object_api.clone(), path_ref, &mode.read_opts()).await?;
+        let (data, obj) = if preserve_empty {
+            read_config_preserve_empty_with_metadata_opts(self.object_api.clone(), path_ref, &mode.read_opts()).await?
+        } else {
+            read_iam_config_with_metadata(self.object_api.clone(), path_ref, &mode.read_opts()).await?
+        };
         let outcome = Self::decrypt_data_with_source(&data)?;
         self.maybe_schedule_lazy_rewrite(path_ref, &outcome, &obj);
 
@@ -614,8 +620,9 @@ impl ObjectStore {
     /// Parameterized core of [`Store::load_user_identity`].
     async fn load_user_identity_with(&self, name: &str, user_type: UserType, mode: LoadMode) -> Result<UserIdentity> {
         let mut u: UserIdentity = self
-            .load_iam_config_with(get_user_identity_path(name, user_type), mode)
+            .load_iamconfig_bytes_with_metadata(get_user_identity_path(name, user_type), mode, true)
             .await
+            .and_then(|(data, _)| serde_json::from_slice(&data).map_err(Error::from))
             .map_err(|err| {
                 if is_err_config_not_found(&err) {
                     debug!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity missing");
@@ -695,7 +702,7 @@ impl ObjectStore {
     /// Parameterized core of [`Store::load_policy`].
     async fn load_policy_with(&self, name: &str, mode: LoadMode) -> Result<PolicyDoc> {
         let (data, obj) = self
-            .load_iamconfig_bytes_with_metadata(get_policy_doc_path(name), mode)
+            .load_iamconfig_bytes_with_metadata(get_policy_doc_path(name), mode, false)
             .await
             .map_err(|err| {
                 if is_err_config_not_found(&err) {

crates/iam/src/sys.rs

@@ -2667,11 +2667,7 @@ mod tests {
                     name: None,
                     description: None,
                 };
-                let parent_identity = UserIdentity {
-                    version: 1,
-                    credentials: creds,
-                    update_at: Some(OffsetDateTime::now_utc()),
-                };
+                let parent_identity = UserIdentity::new(creds);
                 let mut users = HashMap::new();
                 users.insert(PARENT_USER.to_string(), parent_identity);
 
@@ -2703,11 +2699,7 @@ mod tests {
                 name: None,
                 description: None,
             };
-            let parent_identity = UserIdentity {
-                version: 1,
-                credentials: creds,
-                update_at: Some(OffsetDateTime::now_utc()),
-            };
+            let parent_identity = UserIdentity::new(creds);
             let mut users = HashMap::new();
             users.insert(PARENT_USER.to_string(), parent_identity);
 
@@ -2820,16 +2812,12 @@ mod tests {
     async fn group_writes_stamped_before_the_cache_load_time_still_publish() {
         let iam_sys = test_iam_sys().await;
         let member = "group-stamp-member";
-        let identity = UserIdentity {
-            version: 1,
-            credentials: Credentials {
-                access_key: member.to_string(),
-                secret_key: "longenoughsecret".to_string(),
-                status: "on".to_string(),
-                ..Default::default()
-            },
-            update_at: Some(OffsetDateTime::now_utc()),
-        };
+        let identity = UserIdentity::new(Credentials {
+            access_key: member.to_string(),
+            secret_key: "longenoughsecret".to_string(),
+            status: "on".to_string(),
+            ..Default::default()
+        });
         iam_sys.store.cache.with_write_lock(|cache| {
             cache.add_or_update_user(member, &identity, OffsetDateTime::now_utc());
             // The startup load publishes every entity with the load time.

crates/iam/tests/ecstore_test_compat/mod.rs

@@ -22,7 +22,11 @@
 #[allow(unused_imports)]
 pub(crate) mod fixture {
     pub(crate) use rustfs_ecstore::api::bucket::migration::try_migrate_iam_config;
+    pub(crate) use rustfs_ecstore::api::config::com::save_config;
+    pub(crate) use rustfs_ecstore::api::global::set_object_store_resolver;
     pub(crate) use rustfs_ecstore::api::layout::SetupType;
+    pub(crate) use rustfs_ecstore::api::runtime::{InstanceContext, first_cluster_node_is_local, object_store_handle};
+    pub(crate) use rustfs_ecstore::api::storage::ECStore;
 
     // `update_erasure_type` is a write-side global facade entry. Its use is
     // restricted to reviewed storage_api boundaries; this test-only module is

crates/iam/tests/iam_bootstrap_no_lock_test.rs

@@ -33,12 +33,21 @@
 
 mod ecstore_test_compat;
 
-use ecstore_test_compat::fixture::{SetupType, update_erasure_type};
+use ecstore_test_compat::fixture::{
+    ECStore, InstanceContext, SetupType, first_cluster_node_is_local, object_store_handle, save_config,
+    set_object_store_resolver, update_erasure_type,
+};
+use rustfs_credentials::{Credentials, IAM_POLICY_CLAIM_NAME_SA};
 use rustfs_iam::cache::Cache;
-use rustfs_iam::store::object::ObjectStore;
-use rustfs_iam::store::{GroupInfo, Store};
+use rustfs_iam::error::Error as IamError;
+use rustfs_iam::manager::{IamCache, IamState};
+use rustfs_iam::store::object::{IAM_CONFIG_POLICIES_PREFIX, IAM_CONFIG_USERS_PREFIX, ObjectStore};
+use rustfs_iam::store::{GroupInfo, Store, UserType};
+use rustfs_policy::auth::UserIdentity;
 use serial_test::serial;
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64};
+use std::sync::{Arc, Mutex, OnceLock, Weak};
 
 const TEST_GROUP: &str = "seq-restart-group";
 const TEST_MEMBERS: [&str; 2] = ["alice", "bob"];
@@ -69,7 +78,8 @@ async fn load_all_bypasses_namespace_lock_quorum() {
             .build()
             .await
             .ecstore;
-        let store = ObjectStore::new(ecstore);
+        let _route_guard = TestStoreRouteGuard::new(&ecstore);
+        let store = ObjectStore::new(ecstore.clone());
 
         // Seed IAM data while namespace locks still work (single-node mode).
         store
@@ -128,3 +138,447 @@ async fn load_all_bypasses_namespace_lock_quorum() {
     })
     .await;
 }
+
+const IDENTITY_PARENT: &str = "fixture-parent";
+const IDENTITY_CHILD: &str = "fixture-service-child";
+
+static TEST_STORE_ROUTE: Mutex<Option<Weak<ECStore>>> = Mutex::new(None);
+static TEST_STORE_RESOLVER: OnceLock<()> = OnceLock::new();
+
+struct TestStoreRouteGuard;
+
+impl TestStoreRouteGuard {
+    fn new(ecstore: &Arc<ECStore>) -> Self {
+        TEST_STORE_RESOLVER.get_or_init(|| {
+            assert!(
+                set_object_store_resolver(Arc::new(|| {
+                    TEST_STORE_ROUTE
+                        .lock()
+                        .expect("lock test store route")
+                        .as_ref()
+                        .and_then(Weak::upgrade)
+                })),
+                "test binary must own its resolver registration"
+            );
+        });
+        let mut route = TEST_STORE_ROUTE.lock().expect("lock test store route");
+        assert!(route.is_none(), "another test store route is active");
+        *route = Some(Arc::downgrade(ecstore));
+        Self
+    }
+}
+
+impl Drop for TestStoreRouteGuard {
+    fn drop(&mut self) {
+        *TEST_STORE_ROUTE.lock().expect("lock test store route for cleanup") = None;
+    }
+}
+
+struct IdentityStoreFixture {
+    previous_store: Option<Arc<ECStore>>,
+    ecstore: Arc<ECStore>,
+    manager: IamCache<ObjectStore>,
+    _route_guard: TestStoreRouteGuard,
+    _temp_dir: tempfile::TempDir,
+}
+
+impl IdentityStoreFixture {
+    async fn new() -> Self {
+        let temp_dir = tempfile::TempDir::with_prefix("rustfs_iam_identity_fixture_").expect("create identity fixture directory");
+        let ctx = Arc::new(InstanceContext::new());
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .base_dir(temp_dir.path())
+            .init_bucket_metadata(false)
+            .instance_ctx(ctx.clone())
+            .build()
+            .await;
+        ctx.set_endpoints(env.endpoint_pools.clone());
+        let ecstore = env.ecstore;
+        let previous_store = object_store_handle();
+        let route_guard = TestStoreRouteGuard::new(&ecstore);
+        let selected = object_store_handle().expect("fixture route selects its ECStore");
+        assert!(Arc::ptr_eq(&selected, &ecstore), "fixture route must select the isolated ECStore");
+        drop(selected);
+        assert!(first_cluster_node_is_local().await, "isolated fixture has a local first endpoint");
+        temp_env::async_with_vars([("RUSTFS_SKIP_BACKGROUND_TASK", Some("1"))], rustfs_iam::build_iam_sys(ecstore.clone()))
+            .await
+            .expect("initialize IAM metadata through production owner");
+        let store = ObjectStore::new(ecstore.clone());
+        let parent = UserIdentity::new(Credentials {
+            access_key: IDENTITY_PARENT.to_string(),
+            secret_key: "fixture-parent-secret".to_string(),
+            expiration: None,
+            session_token: String::new(),
+            status: "on".to_string(),
+            ..Default::default()
+        });
+        let child = UserIdentity::new(Credentials {
+            access_key: IDENTITY_CHILD.to_string(),
+            secret_key: "fixture-child-secret".to_string(),
+            expiration: None,
+            session_token: String::new(),
+            status: "on".to_string(),
+            parent_user: IDENTITY_PARENT.to_string(),
+            claims: Some(HashMap::from([(
+                IAM_POLICY_CLAIM_NAME_SA.to_string(),
+                serde_json::json!("inherited-policy"),
+            )])),
+            ..Default::default()
+        });
+        assert!(child.credentials.is_service_account(), "child fixture must be a service account");
+        store
+            .save_user_identity(IDENTITY_PARENT, UserType::Reg, parent, None)
+            .await
+            .expect("persist parent identity");
+        store
+            .save_user_identity(IDENTITY_CHILD, UserType::Svc, child, None)
+            .await
+            .expect("persist service child identity");
+        let manager = IamCache {
+            cache: Cache::default(),
+            api: store,
+            state: Arc::new(AtomicU8::new(IamState::Ready as u8)),
+            loading: Arc::new(AtomicBool::new(false)),
+            roles: HashMap::new(),
+            send_chan: tokio::sync::mpsc::channel::<i64>(1).0,
+            last_timestamp: AtomicI64::new(0),
+            sync_failures: AtomicU64::new(0),
+            sync_successes: AtomicU64::new(0),
+            last_sync_duration_millis: AtomicU64::new(0),
+        };
+        manager
+            .load_user(IDENTITY_PARENT)
+            .await
+            .expect("load persisted parent into cache");
+        manager
+            .load_user(IDENTITY_CHILD)
+            .await
+            .expect("load persisted service child into cache");
+        Self {
+            previous_store,
+            ecstore,
+            manager,
+            _route_guard: route_guard,
+            _temp_dir: temp_dir,
+        }
+    }
+
+    async fn assert_parent_and_child(&self) {
+        let parent = self
+            .manager
+            .api
+            .load_user_identity(IDENTITY_PARENT, UserType::Reg)
+            .await
+            .expect("read parent from real storage");
+        assert_eq!(parent.credentials.access_key, IDENTITY_PARENT);
+        assert_eq!(parent.credentials.parent_user, "");
+        let child = self
+            .manager
+            .api
+            .load_user_identity(IDENTITY_CHILD, UserType::Svc)
+            .await
+            .expect("read child from real storage");
+        assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
+        assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
+        assert!(child.credentials.is_service_account());
+        let cached_parent = self.manager.get_user(IDENTITY_PARENT).await.expect("parent remains in cache");
+        assert_eq!(cached_parent.credentials.access_key, IDENTITY_PARENT);
+        assert_eq!(cached_parent.credentials.parent_user, "");
+        let cached_child = self
+            .manager
+            .get_user(IDENTITY_CHILD)
+            .await
+            .expect("service child remains in cache");
+        assert_eq!(cached_child.credentials.access_key, IDENTITY_CHILD);
+        assert_eq!(cached_child.credentials.parent_user, IDENTITY_PARENT);
+        assert!(cached_child.credentials.is_service_account());
+    }
+
+    async fn overwrite_parent_bytes(&self, bytes: &[u8]) {
+        let path = format!("{}{IDENTITY_PARENT}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
+        save_config(self.ecstore.clone(), &path, bytes.to_vec())
+            .await
+            .expect("overwrite real parent identity bytes");
+    }
+
+    async fn assert_child_storage_and_identity_cache(&self) {
+        let child = self
+            .manager
+            .api
+            .load_user_identity(IDENTITY_CHILD, UserType::Svc)
+            .await
+            .expect("notification must preserve persisted service child");
+        assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
+        assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
+        assert!(child.credentials.is_service_account());
+        let parent = self
+            .manager
+            .get_user(IDENTITY_PARENT)
+            .await
+            .expect("failed notification preserves cached parent");
+        assert_eq!(parent.credentials.access_key, IDENTITY_PARENT);
+        assert_eq!(parent.credentials.parent_user, "");
+        let child = self
+            .manager
+            .get_user(IDENTITY_CHILD)
+            .await
+            .expect("failed notification preserves cached child");
+        assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
+        assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
+        assert!(child.credentials.is_service_account());
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_store_fixture_roundtrips_parent_and_service_child() {
+    let fixture = IdentityStoreFixture::new().await;
+    let before = fixture.previous_store.clone();
+    let selected = object_store_handle().expect("fixture route selects its ECStore");
+    assert!(Arc::ptr_eq(&selected, &fixture.ecstore));
+    drop(selected);
+    fixture.assert_parent_and_child().await;
+    drop(fixture);
+    assert!(TEST_STORE_ROUTE.lock().expect("inspect cleared test store route").is_none());
+    match (before, object_store_handle()) {
+        (None, None) => {}
+        (Some(before), Some(after)) => assert!(Arc::ptr_eq(&before, &after), "fixture drop restores prior store identity"),
+        _ => panic!("fixture drop must restore the prior store option"),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_notification_valid_parent_preserves_child_storage_and_cache() {
+    let fixture = IdentityStoreFixture::new().await;
+    fixture.assert_parent_and_child().await;
+    fixture
+        .manager
+        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
+        .await
+        .expect("refresh valid parent notification");
+    fixture.assert_parent_and_child().await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_store_fixture_raw_valid_overwrite_is_observed() {
+    let fixture = IdentityStoreFixture::new().await;
+    let parent = fixture
+        .manager
+        .api
+        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
+        .await
+        .expect("load valid parent before raw overwrite");
+    let mut raw = serde_json::to_value(parent).expect("serialize valid parent fixture");
+    raw["credentials"]["name"] = serde_json::json!("updated-name");
+    let path = format!("{}{IDENTITY_PARENT}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
+    save_config(
+        fixture.ecstore.clone(),
+        &path,
+        serde_json::to_vec(&raw).expect("encode raw parent fixture"),
+    )
+    .await
+    .expect("overwrite parent through real config storage");
+    fixture
+        .manager
+        .load_user(IDENTITY_PARENT)
+        .await
+        .expect("reload parent after raw overwrite");
+    let stored = fixture
+        .manager
+        .api
+        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
+        .await
+        .expect("read overwritten parent from storage");
+    assert_eq!(stored.credentials.name.as_deref(), Some("updated-name"));
+    let cached = fixture
+        .manager
+        .get_user(IDENTITY_PARENT)
+        .await
+        .expect("read overwritten parent from cache");
+    assert_eq!(cached.credentials.name.as_deref(), Some("updated-name"));
+    fixture.assert_parent_and_child().await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_corrupt_reads_preserve_crypto_error() {
+    let fixture = IdentityStoreFixture::new().await;
+    fixture.overwrite_parent_bytes(&[0xff, 0xfe, 0xfd]).await;
+    let locked = fixture
+        .manager
+        .api
+        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
+        .await
+        .expect_err("corrupt stored identity must fail locked read");
+    assert!(matches!(locked, IamError::CryptoError(_)), "preserve crypto error, got {locked:?}");
+    let mut users = HashMap::new();
+    let no_lock = fixture
+        .manager
+        .api
+        .load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
+        .await
+        .expect_err("corrupt stored identity must fail no-lock read");
+    assert!(matches!(no_lock, IamError::CryptoError(_)), "preserve crypto error, got {no_lock:?}");
+    assert!(users.is_empty(), "failed no-lock read must not populate users");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_corrupt_notification_preserves_child_storage_and_cache() {
+    let fixture = IdentityStoreFixture::new().await;
+    fixture.assert_parent_and_child().await;
+    fixture.overwrite_parent_bytes(&[0xff, 0xfe, 0xfd]).await;
+    let notification = fixture
+        .manager
+        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
+        .await;
+    fixture.assert_child_storage_and_identity_cache().await;
+    let error = notification.expect_err("corrupt parent notification must return an error");
+    assert!(
+        matches!(error, IamError::CryptoError(_)),
+        "preserve notification crypto error, got {error:?}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_invalid_schema_preserves_decode_error() {
+    let fixture = IdentityStoreFixture::new().await;
+    fixture.assert_parent_and_child().await;
+    fixture
+        .overwrite_parent_bytes(br#"{"version":"not-an-integer","credentials":{}}"#)
+        .await;
+    let error = fixture
+        .manager
+        .api
+        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
+        .await
+        .expect_err("invalid identity field type must fail decode");
+    assert_identity_data_error(error);
+    let mut users = HashMap::new();
+    let error = fixture
+        .manager
+        .api
+        .load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
+        .await
+        .expect_err("invalid identity field type must fail no-lock decode");
+    assert_identity_data_error(error);
+    assert!(users.is_empty(), "failed decode must not populate users");
+    let notification = fixture
+        .manager
+        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
+        .await;
+    fixture.assert_child_storage_and_identity_cache().await;
+    assert_identity_data_error(notification.expect_err("invalid parent notification must return decode error"));
+}
+
+fn assert_identity_data_error(error: IamError) {
+    let IamError::Io(inner) = error else {
+        panic!("identity decode must preserve Io serde error, got {error:?}");
+    };
+    let cause = inner
+        .get_ref()
+        .and_then(|cause| cause.downcast_ref::<serde_json::Error>())
+        .expect("identity decode error retains serde source");
+    assert!(cause.is_data(), "identity field type must produce serde Data error");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_physically_missing_is_no_such_user() {
+    let fixture = IdentityStoreFixture::new().await;
+    let name = "physically-missing-identity";
+    let error = fixture
+        .manager
+        .api
+        .load_user_identity(name, UserType::Reg)
+        .await
+        .expect_err("missing identity must be absent");
+    assert!(
+        matches!(error, IamError::NoSuchUser(ref missing) if missing == name),
+        "preserve absent name, got {error:?}"
+    );
+    let mut users = HashMap::new();
+    let error = fixture
+        .manager
+        .api
+        .load_user_no_lock(name, UserType::Reg, &mut users)
+        .await
+        .expect_err("missing no-lock identity must be absent");
+    assert!(
+        matches!(error, IamError::NoSuchUser(ref missing) if missing == name),
+        "preserve absent name, got {error:?}"
+    );
+    assert!(users.is_empty(), "missing read must not populate users");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_empty_reads_preserve_crypto_error() {
+    let fixture = IdentityStoreFixture::new().await;
+    fixture.overwrite_parent_bytes(&[]).await;
+    let locked = fixture
+        .manager
+        .api
+        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
+        .await
+        .expect_err("empty stored identity must fail locked read");
+    assert!(matches!(locked, IamError::CryptoError(_)), "preserve crypto error, got {locked:?}");
+    let mut users = HashMap::new();
+    let no_lock = fixture
+        .manager
+        .api
+        .load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
+        .await
+        .expect_err("empty stored identity must fail no-lock read");
+    assert!(matches!(no_lock, IamError::CryptoError(_)), "preserve crypto error, got {no_lock:?}");
+    assert!(users.is_empty(), "failed no-lock read must not populate users");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn identity_empty_notification_preserves_child_storage_and_cache() {
+    let fixture = IdentityStoreFixture::new().await;
+    fixture.assert_parent_and_child().await;
+    fixture.overwrite_parent_bytes(&[]).await;
+    let notification = fixture
+        .manager
+        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
+        .await;
+    fixture.assert_child_storage_and_identity_cache().await;
+    let error = notification.expect_err("empty parent notification must return an error");
+    assert!(
+        matches!(error, IamError::CryptoError(_)),
+        "preserve notification crypto error, got {error:?}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn policy_empty_payload_retains_no_such_policy() {
+    let fixture = IdentityStoreFixture::new().await;
+    let name = "empty-policy";
+    let path = format!("{}{name}/policy.json", IAM_CONFIG_POLICIES_PREFIX.as_str());
+    save_config(fixture.ecstore.clone(), &path, Vec::new())
+        .await
+        .expect("save real empty policy payload");
+    let locked = fixture
+        .manager
+        .api
+        .load_policy(name)
+        .await
+        .err()
+        .expect("empty policy must retain absent-policy behavior");
+    assert!(matches!(locked, IamError::NoSuchPolicy), "preserve policy error, got {locked:?}");
+    let mut policies = HashMap::new();
+    let no_lock = fixture
+        .manager
+        .api
+        .load_policy_doc_no_lock(name, &mut policies)
+        .await
+        .expect_err("empty no-lock policy must retain absent-policy behavior");
+    assert!(matches!(no_lock, IamError::NoSuchPolicy), "preserve policy error, got {no_lock:?}");
+    assert!(policies.is_empty(), "failed no-lock policy read must not populate policies");
+}

crates/policy/Cargo.toml

@@ -58,6 +58,7 @@ tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
 time = { workspace = true, features = ["serde", "parsing", "formatting", "macros"] }
 serde = { workspace = true, features = ["derive", "rc"] }
 serde_json = { workspace = true, features = ["raw_value"] }
+uuid = { workspace = true, features = ["serde"] }
 thiserror.workspace = true
 strum = { workspace = true, features = ["derive"] }
 rustfs-crypto = { workspace = true }

crates/policy/src/auth/mod.rs

@@ -22,6 +22,33 @@ use serde_json::json;
 use std::collections::HashMap;
 use time::OffsetDateTime;
 
+/// A nonnil stored principal identifier; parsing it does not establish identity authority.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(try_from = "uuid::Uuid", into = "uuid::Uuid")]
+pub struct PrincipalIncarnation(uuid::Uuid);
+
+#[derive(Debug, thiserror::Error)]
+#[error("principal incarnation must not be nil")]
+pub struct InvalidPrincipalIncarnation;
+
+impl TryFrom<uuid::Uuid> for PrincipalIncarnation {
+    type Error = InvalidPrincipalIncarnation;
+
+    fn try_from(value: uuid::Uuid) -> Result<Self, Self::Error> {
+        if value.is_nil() {
+            Err(InvalidPrincipalIncarnation)
+        } else {
+            Ok(Self(value))
+        }
+    }
+}
+
+impl From<PrincipalIncarnation> for uuid::Uuid {
+    fn from(value: PrincipalIncarnation) -> Self {
+        value.0
+    }
+}
+
 #[derive(Serialize, Deserialize, Clone, Debug, Default)]
 pub struct UserIdentity {
     pub version: i64,
@@ -29,6 +56,8 @@ pub struct UserIdentity {
     /// updatedAt (RFC3339), legacy RustFS: update_at. Serialize as updatedAt
     #[serde(rename = "updatedAt", alias = "update_at", default, with = "crate::serde_datetime::option")]
     pub update_at: Option<OffsetDateTime>,
+    #[serde(default, rename = "principalIncarnation", skip_serializing_if = "Option::is_none")]
+    pub principal_incarnation: Option<PrincipalIncarnation>,
 }
 
 impl UserIdentity {
@@ -44,6 +73,7 @@ impl UserIdentity {
             version: 1,
             credentials,
             update_at: Some(OffsetDateTime::now_utc()),
+            principal_incarnation: None,
         }
     }
 
@@ -73,13 +103,124 @@ impl From<Credentials> for UserIdentity {
             version: 1,
             credentials: value,
             update_at: Some(OffsetDateTime::now_utc()),
+            principal_incarnation: None,
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::UserIdentity;
+    use super::{PrincipalIncarnation, UserIdentity};
+    use rustfs_credentials::Credentials;
+    use serde_json::json;
+    use uuid::Uuid;
+
+    const INCARNATION: &str = "8f422eba-aab4-4ec3-a7ec-d1717463b520";
+
+    #[test]
+    fn test_user_identity_legacy_fields_absent() {
+        let mut legacy = json!({"version": 1, "credentials": {"accessKey": "ak", "secretKey": "sk12345678"}});
+        let identity: UserIdentity = serde_json::from_value(legacy.clone()).expect("decode legacy identity");
+        assert!(identity.principal_incarnation.is_none());
+        legacy["principalIncarnation"] = serde_json::Value::Null;
+        let identity: UserIdentity = serde_json::from_value(legacy).expect("decode null incarnation");
+        assert!(identity.principal_incarnation.is_none());
+    }
+
+    #[test]
+    fn test_user_identity_legacy_serialization_shape() {
+        let identity = UserIdentity::default();
+        let serialized = serde_json::to_value(&identity).expect("serialize legacy identity");
+        assert_eq!(serialized, json!({"version": 0, "credentials": identity.credentials, "updatedAt": null}));
+        assert!(serialized.get("principalIncarnation").is_none());
+    }
+
+    #[test]
+    fn test_user_identity_constructors_do_not_mint_incarnation() {
+        let default = UserIdentity::default();
+        assert!(default.principal_incarnation.is_none());
+        assert_eq!(default.version, 0);
+        assert!(default.update_at.is_none());
+        let credentials = Credentials {
+            access_key: "ak".to_string(),
+            secret_key: "sk12345678".to_string(),
+            ..Default::default()
+        };
+        let expected_credentials = serde_json::to_value(&credentials).expect("serialize expected credentials");
+        for identity in [UserIdentity::new(credentials.clone()), UserIdentity::from(credentials)] {
+            assert!(identity.principal_incarnation.is_none());
+            assert_eq!(identity.version, 1);
+            assert!(identity.update_at.is_some());
+            assert_eq!(
+                serde_json::to_value(&identity.credentials).expect("serialize constructor credentials"),
+                expected_credentials
+            );
+            assert!(
+                serde_json::to_value(identity)
+                    .expect("serialize constructor identity")
+                    .get("principalIncarnation")
+                    .is_none()
+            );
+        }
+    }
+
+    #[test]
+    fn test_principal_incarnation_rejects_nil_and_malformed() {
+        assert!(PrincipalIncarnation::try_from(Uuid::nil()).is_err());
+        for value in ["00000000-0000-0000-0000-000000000000", "not-a-uuid", ""] {
+            assert!(serde_json::from_value::<PrincipalIncarnation>(json!(value)).is_err(), "reject {value}");
+            let identity = json!({"version": 1, "credentials": {}, "principalIncarnation": value});
+            assert!(serde_json::from_value::<UserIdentity>(identity).is_err(), "identity rejects {value}");
+        }
+    }
+
+    #[test]
+    fn test_principal_incarnation_roundtrip_preserves_uuid() {
+        let uuid = Uuid::parse_str(INCARNATION).expect("parse fixed incarnation");
+        let incarnation = PrincipalIncarnation::try_from(uuid).expect("accept nonnil incarnation");
+        assert_eq!(Uuid::from(incarnation).as_bytes(), uuid.as_bytes());
+        let serialized = serde_json::to_value(incarnation).expect("serialize incarnation");
+        assert_eq!(serialized, json!(INCARNATION));
+        assert_eq!(
+            serde_json::from_value::<PrincipalIncarnation>(serialized).expect("decode incarnation"),
+            incarnation
+        );
+    }
+
+    #[test]
+    fn test_user_identity_live_incarnation_roundtrip() {
+        let mut identity = UserIdentity::new(Credentials::default());
+        identity.principal_incarnation = Some(
+            PrincipalIncarnation::try_from(Uuid::parse_str(INCARNATION).expect("parse fixed incarnation"))
+                .expect("accept fixed incarnation"),
+        );
+        let serialized = serde_json::to_value(&identity).expect("serialize identity with incarnation");
+        assert_eq!(serialized["principalIncarnation"], INCARNATION);
+        let encoded = serde_json::to_string(&identity).expect("encode identity with incarnation");
+        let decoded: UserIdentity = serde_json::from_str(&encoded).expect("decode identity with incarnation");
+        assert_eq!(decoded.principal_incarnation, identity.principal_incarnation);
+        assert_eq!(serde_json::to_value(decoded).expect("reserialize identity"), serialized);
+    }
+
+    #[test]
+    fn test_user_identity_timestamp_aliases_unchanged() {
+        let mut minio = json!({"version": 1, "credentials": {}, "updatedAt": "2025-03-07T12:00:00Z"});
+        let encoded = serde_json::to_string(&minio).expect("encode updatedAt fixture");
+        let identity: UserIdentity = serde_json::from_str(&encoded).expect("decode updatedAt timestamp");
+        minio["update_at"] = minio
+            .as_object_mut()
+            .expect("identity object")
+            .remove("updatedAt")
+            .expect("existing timestamp");
+        let encoded = serde_json::to_string(&minio).expect("encode update_at fixture");
+        let legacy: UserIdentity = serde_json::from_str(&encoded).expect("decode update_at alias");
+        assert_eq!(legacy.update_at, identity.update_at);
+        assert_eq!(legacy.version, 1);
+        let serialized = serde_json::to_value(legacy).expect("serialize legacy timestamp");
+        assert_eq!(serialized["updatedAt"], "2025-03-07T12:00:00Z");
+        assert!(serialized.get("update_at").is_none());
+        assert!(serialized.get("principalIncarnation").is_none());
+    }
 
     /// Deserialize UserIdentity from MinIO-style JSON (RFC3339 updatedAt).
     #[test]

crates/test-utils/src/ecstore_test_compat.rs

@@ -27,9 +27,10 @@ pub(crate) mod fixture {
     pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint;
     pub(crate) use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints};
     pub(crate) use rustfs_ecstore::api::object::{PutObjReader, SelectObjectSnapshot};
+    pub(crate) use rustfs_ecstore::api::runtime::InstanceContext;
     #[cfg(feature = "put-object-commit-barrier")]
     pub(crate) use rustfs_ecstore::api::set_disk as ecstore_set_disk;
-    pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks};
+    pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks, init_local_disks_with_instance_ctx};
     pub(crate) use rustfs_storage_api::{BucketOperations, BucketOptions, MakeBucketOptions, ObjectIO};
 
     #[cfg(test)]

crates/test-utils/src/lib.rs

@@ -35,8 +35,9 @@ use std::sync::{Arc, Once};
 #[cfg(feature = "put-object-commit-barrier")]
 use ecstore_test_compat::fixture::ecstore_set_disk;
 use ecstore_test_compat::fixture::{
-    BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _,
-    PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks,
+    BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, MakeBucketOptions,
+    ObjectIO as _, PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks,
+    init_local_disks_with_instance_ctx,
 };
 use tokio_util::sync::CancellationToken;
 
@@ -90,7 +91,7 @@ pub struct TestECStoreEnv {
     pub temp_root: PathBuf,
     /// The per-disk directories (`disk1`..`diskN`) under `temp_root`.
     pub disk_paths: Vec<PathBuf>,
-    /// The store, bootstrapped exactly like the historical test setups:
+    /// By default, the store is bootstrapped exactly like the historical test setups:
     /// `init_local_disks` + `ECStore::new` on `127.0.0.1:0` (random port keeps
     /// nextest's process-per-test parallelism safe).
     pub ecstore: Arc<ECStore>,
@@ -155,6 +156,7 @@ pub struct TestECStoreEnvBuilder {
     prefix: String,
     base_dir: Option<PathBuf>,
     init_bucket_metadata: bool,
+    instance_ctx: Option<Arc<InstanceContext>>,
 }
 
 impl Default for TestECStoreEnvBuilder {
@@ -164,6 +166,7 @@ impl Default for TestECStoreEnvBuilder {
             prefix: "rustfs_test_utils".to_string(),
             base_dir: None,
             init_bucket_metadata: true,
+            instance_ctx: None,
         }
     }
 }
@@ -196,6 +199,12 @@ impl TestECStoreEnvBuilder {
         self
     }
 
+    /// Use a caller-owned context for local disks and store construction.
+    pub fn instance_ctx(mut self, instance_ctx: Arc<InstanceContext>) -> Self {
+        self.instance_ctx = Some(instance_ctx);
+        self
+    }
+
     /// Build the environment. Panics on any bootstrap failure — this is test
     /// scaffolding, and a broken environment must fail the test loudly.
     pub async fn build(self) -> TestECStoreEnv {
@@ -237,14 +246,22 @@ impl TestECStoreEnvBuilder {
         };
         let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]);
 
-        init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
-
         // Port 0 keeps ECStore-backed integration binaries parallel-safe under
         // nextest: no fixed peer port is ever shared between test processes.
         let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().expect("parse test addr");
-        let ecstore = ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new())
-            .await
-            .expect("build test ECStore");
+        let ecstore = if let Some(instance_ctx) = self.instance_ctx {
+            init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
+                .await
+                .expect("init instance local disks");
+            ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), CancellationToken::new(), instance_ctx)
+                .await
+                .expect("build instance test ECStore")
+        } else {
+            init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
+            ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new())
+                .await
+                .expect("build test ECStore")
+        };
 
         // The production bootstrap only persists pool.bin from the elected
         // first cluster node.  Test stores intentionally have no cluster
@@ -278,3 +295,86 @@ impl TestECStoreEnvBuilder {
         }
     }
 }
+
+#[cfg(test)]
+mod context_tests {
+    use super::{InstanceContext, TestECStoreEnv};
+    use crate::ecstore_test_compat::fixture::{read_config, save_config};
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn explicit_context_keeps_stores_isolated() {
+        let root_a = tempfile::tempdir().expect("create context A disk root");
+        let root_b = tempfile::tempdir().expect("create context B disk root");
+        let ctx_a = Arc::new(InstanceContext::new());
+        let ctx_b = Arc::new(InstanceContext::new());
+        let env_a = TestECStoreEnv::builder()
+            .base_dir(root_a.path())
+            .init_bucket_metadata(false)
+            .instance_ctx(ctx_a.clone())
+            .build()
+            .await;
+        let env_b = TestECStoreEnv::builder()
+            .base_dir(root_b.path())
+            .init_bucket_metadata(false)
+            .instance_ctx(ctx_b.clone())
+            .build()
+            .await;
+
+        assert!(env_a.ecstore.instance_endpoints().is_none());
+        assert!(env_b.ecstore.instance_endpoints().is_none());
+        ctx_a.set_endpoints(env_a.endpoint_pools.clone());
+        let endpoints_a = env_a
+            .ecstore
+            .instance_endpoints()
+            .expect("instance A sees its caller-owned context topology");
+        assert_eq!(
+            endpoints_a.0[0]
+                .endpoints
+                .into_ref()
+                .first()
+                .expect("context A first endpoint")
+                .to_string(),
+            env_a.disk_paths[0].to_string_lossy(),
+        );
+        assert!(
+            env_b.ecstore.instance_endpoints().is_none(),
+            "publishing context A must not publish context B"
+        );
+        ctx_b.set_endpoints(env_b.endpoint_pools.clone());
+        let endpoints_b = env_b
+            .ecstore
+            .instance_endpoints()
+            .expect("instance B sees its caller-owned context topology");
+        assert_eq!(
+            endpoints_b.0[0]
+                .endpoints
+                .into_ref()
+                .first()
+                .expect("context B first endpoint")
+                .to_string(),
+            env_b.disk_paths[0].to_string_lossy(),
+        );
+        assert_eq!(
+            env_a.ecstore.instance_endpoints().expect("context A remains published").0[0]
+                .endpoints
+                .into_ref()
+                .first()
+                .expect("context A retained endpoint")
+                .to_string(),
+            env_a.disk_paths[0].to_string_lossy(),
+        );
+
+        let path = "config/test-context-isolation.json";
+        let bytes_a = br#"{"context":"a"}"#.to_vec();
+        let bytes_b = br#"{"context":"b"}"#.to_vec();
+        save_config(env_a.ecstore.clone(), path, bytes_a.clone())
+            .await
+            .expect("write context A config");
+        save_config(env_b.ecstore.clone(), path, bytes_b.clone())
+            .await
+            .expect("write context B config");
+        assert_eq!(read_config(env_a.ecstore.clone(), path).await.expect("read context A config"), bytes_a);
+        assert_eq!(read_config(env_b.ecstore.clone(), path).await.expect("read context B config"), bytes_b);
+    }
+}

scripts/tests/test_capture_dependencies.py

@@ -0,0 +1,154 @@
+"""Verify the real locked capture graph and leader identity feature contract."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import re
+import unittest
+
+ROOT = Path(__file__).resolve().parents[2]
+REGISTRY = "registry+https://github.com/rust-lang/crates.io-index"
+
+
+def source_files(root):
+    result = subprocess.run(["git", "ls-files", "-z"], cwd=root, check=True, capture_output=True)
+    paths = {Path(os.fsdecode(p)) for p in result.stdout.split(b"\0") if p}
+    paths.add(Path("scripts/tests/test_capture_dependencies.py"))
+    return sorted(paths)
+
+
+def fingerprint(root, paths):
+    digest = hashlib.sha256()
+    for path in paths:
+        digest.update(os.fsencode(path))
+        digest.update(b"\0")
+        source = root / path
+        if source.is_symlink():
+            digest.update(b"symlink\0")
+            digest.update(os.fsencode(os.readlink(source)))
+        else:
+            digest.update(b"file\0")
+            digest.update(source.read_bytes())
+    return digest.hexdigest()
+
+
+def metadata(root, all_features=False):
+    lock = (root / "Cargo.lock").read_bytes()
+    command = ["cargo", "metadata", "--locked", "--format-version", "1"]
+    if all_features:
+        command.append("--all-features")
+    result = subprocess.run(command, cwd=root, capture_output=True, text=True)
+    if result.returncode:
+        raise RuntimeError(f"{' '.join(command)} exited {result.returncode}: {result.stderr}")
+    if (root / "Cargo.lock").read_bytes() != lock:
+        raise AssertionError("CAPTURE_LOCK_CHANGED")
+    print(f"{' '.join(command)}: exit 0; lock {hashlib.sha256(lock).hexdigest()}")
+    return json.loads(result.stdout)
+
+
+def assert_graph(case, root, graph):
+    packages = {package["id"]: package for package in graph["packages"]}
+    capture = next(p for p in packages.values() if p["name"] == "rustfs-capture-authority")
+    dependencies = {p["name"]: p for p in capture["dependencies"]}
+    case.assertIn("openraft", dependencies, "CAPTURE_OPENRAFT_REQUIRED")
+    dependency = dependencies["openraft"]
+    case.assertEqual(dependency["req"], "=0.9.25", "CAPTURE_OPENRAFT_EXACT_PIN")
+    case.assertFalse(dependency["uses_default_features"], "CAPTURE_OPENRAFT_DEFAULTS_DISABLED")
+    case.assertIn("serde", dependency["features"])
+    case.assertEqual(dependency["source"], REGISTRY)
+    for name, kind in (("serde", None), ("url", None), ("serde_json", "dev")):
+        case.assertIn(name, dependencies)
+        case.assertEqual(dependencies[name]["kind"], kind)
+    case.assertIn("derive", dependencies["serde"]["features"])
+    nodes = {node["id"]: node for node in graph["resolve"]["nodes"]}
+    reachable = set()
+    pending = [capture["id"]]
+    while pending:
+        node_id = pending.pop()
+        if node_id not in reachable:
+            reachable.add(node_id)
+            pending.extend(nodes[node_id]["dependencies"])
+    openraft = [p for p in packages.values() if p["name"] == "openraft"]
+    case.assertEqual(len(openraft), 1, "CAPTURE_UNIQUE_OPENRAFT")
+    package = openraft[0]
+    case.assertIn(package["id"], reachable)
+    case.assertEqual(package["version"], "0.9.25")
+    case.assertEqual(package["source"], REGISTRY)
+    features = set(nodes[package["id"]]["features"])
+    case.assertNotIn("single-term-leader", features, "CAPTURE_ADVANCED_LEADER_REQUIRED")
+    case.assertNotIn("storage-v2", features, "CAPTURE_STORAGE_V1_REQUIRED")
+    case.assertIn("serde", features)
+    case.assertEqual(set(dependency["features"]), {"serde"}, "CAPTURE_OPENRAFT_SERDE_ONLY")
+    for name in ("serde", "serde_derive", "url", "serde_json"):
+        case.assertTrue(any(packages[p]["name"] == name for p in reachable), name)
+    case.assertTrue(any("derive" in nodes[p]["features"] for p in reachable if packages[p]["name"] == "serde"))
+    blocks = (root / "Cargo.lock").read_text().split("[[package]]")
+    locked = [dict(re.findall(r'^([a-z]+) = "([^"\n]*)"$', block, re.MULTILINE))
+              for block in blocks if re.search(r'^name = "openraft"$', block, re.MULTILINE)]
+    case.assertEqual(len(locked), 1)
+    case.assertEqual(locked[0]["version"], "0.9.25")
+    case.assertEqual(locked[0]["source"], REGISTRY)
+    case.assertEqual(locked[0]["checksum"], "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd", "CAPTURE_PUBLISHED_OPENRAFT_CHECKSUM")
+
+
+class CaptureDependencies(unittest.TestCase):
+    def test_accepted_dependency_graph(self):
+        for all_features in (False, True):
+            with self.subTest(all_features=all_features):
+                graph = metadata(ROOT, all_features)
+                assert_graph(self, ROOT, graph)
+                package = next(p for p in graph["packages"] if p["name"] == "openraft")
+                node = next(n for n in graph["resolve"]["nodes"] if n["id"] == package["id"])
+                print(f"accepted all_features={all_features}: {package['id']}; features {node['features']}")
+
+    def control(self, old, new, expected):
+        paths = source_files(ROOT)
+        before = fingerprint(ROOT, paths)
+        with tempfile.TemporaryDirectory(prefix="capture-dependencies-") as directory:
+            copied = Path(directory)
+            for path in paths:
+                destination = copied / path
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                source = ROOT / path
+                if source.is_symlink():
+                    destination.symlink_to(os.readlink(source))
+                else:
+                    shutil.copyfile(source, destination)
+            self.assertEqual(fingerprint(copied, paths), before)
+            manifest = copied / "Cargo.toml"
+            original = manifest.read_bytes()
+            self.assertEqual(original.count(old.encode()), 1)
+            try:
+                manifest.write_bytes(original.replace(old.encode(), new.encode(), 1))
+                mutated = fingerprint(copied, paths)
+                self.assertNotEqual(mutated, before)
+                for all_features in (False, True):
+                    graph = metadata(copied, all_features)
+                    versions = {p["version"] for p in graph["packages"] if p["name"] == "openraft"}
+                    self.assertEqual(versions, {"0.9.25"})
+                    with self.assertRaisesRegex(AssertionError, expected):
+                        assert_graph(self, copied, graph)
+                print(f"{expected}: metadata exit 0 (default/all-features); source {before}; control {mutated}")
+            finally:
+                manifest.write_bytes(original)
+                self.assertEqual(fingerprint(copied, paths), before)
+                self.assertEqual(fingerprint(ROOT, paths), before)
+                print(f"{expected}: restored source {before}")
+
+    def test_relaxed_pin_is_rejected(self):
+        self.control('version = "=0.9.25"', 'version = "0.9.25"', "CAPTURE_OPENRAFT_EXACT_PIN")
+
+    def test_single_term_leader_is_rejected(self):
+        self.control(
+            'openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }',
+            'openraft = { version = "=0.9.25", default-features = false, features = ["serde", "single-term-leader"] }',
+            "CAPTURE_ADVANCED_LEADER_REQUIRED",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()