Implement UpdateCollection for consensus state machine (#10403)

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

lib/storage/src/content_manager/collection_meta_ops.rs

@@ -367,7 +367,7 @@ pub struct UpdateCollection {
 pub struct UpdateCollectionOperation {
     pub collection_name: String,
     pub update_collection: UpdateCollection,
-    shard_replica_changes: Option<Vec<replica_set::Change>>,
+    pub shard_replica_changes: Option<Vec<replica_set::Change>>,
 }
 
 impl UpdateCollectionOperation {
@@ -404,6 +404,10 @@ impl UpdateCollectionOperation {
         })
     }
 
+    pub fn has_shard_replica_changes(&self) -> bool {
+        self.shard_replica_changes.is_some()
+    }
+
     pub fn take_shard_replica_changes(&mut self) -> Option<Vec<replica_set::Change>> {
         self.shard_replica_changes.take()
     }

lib/storage/src/content_manager/consensus_state_machine/action.rs

@@ -1,14 +1,23 @@
 use std::collections::{BTreeMap, BTreeSet};
 
 use collection::collection_state;
-use collection::operations::types::PeerMetadata;
+use collection::config::CollectionConfigInternal;
+use collection::operations::config_diff::{
+    CollectionParamsDiff, DiffConfig as _, HnswConfigDiff, OptimizersConfigDiff,
+    QuantizationConfigDiff,
+};
+use collection::operations::types::{PeerMetadata, SparseVectorsConfig, VectorsConfigDiff};
 use collection::shards::CollectionId;
 use collection::shards::shard::PeerId;
-use segment::types::{PayloadFieldSchema, PayloadKeyType, VectorNameBuf};
+use segment::types::{
+    Payload, PayloadFieldSchema, PayloadKeyType, QuantizationConfig, StrictModeConfig,
+    VectorNameBuf,
+};
 use shard::operations::vector_name_ops::VectorNameConfig;
 
 #[cfg(feature = "staging")]
 use crate::content_manager::collection_meta_ops::{TestSlowDown, TestTransientError};
+use crate::content_manager::errors::StorageResult;
 use crate::quota::QuotaConfig;
 
 /// A single change a consensus operation makes
@@ -23,6 +32,11 @@ pub enum Action {
         collection: CollectionId,
     },
 
+    UpdateCollectionConfig {
+        collection: CollectionId,
+        diff: Box<CollectionConfigDiff>,
+    },
+
     AddNamedVector {
         collection: CollectionId,
         vector_name: VectorNameBuf,
@@ -79,6 +93,7 @@ impl Action {
         match self {
             Action::CreateCollection { collection, .. }
             | Action::DropCollection { collection }
+            | Action::UpdateCollectionConfig { collection, .. }
             | Action::AddNamedVector { collection, .. }
             | Action::DropNamedVector { collection, .. }
             | Action::SetPayloadIndex { collection, .. }
@@ -95,3 +110,79 @@ impl Action {
         }
     }
 }
+
+/// One of the config updates `UpdateCollection` makes, each a separate save today
+#[derive(Clone, Debug, PartialEq)]
+pub enum CollectionConfigDiff {
+    Optimizers(OptimizersConfigDiff),
+    Params(CollectionParamsDiff),
+    Hnsw(HnswConfigDiff),
+    Vectors(VectorsConfigDiff),
+    Quantization(QuantizationConfigDiff),
+    SparseVectors(SparseVectorsConfig),
+    StrictMode(StrictModeConfig),
+    Metadata(Payload),
+}
+
+impl CollectionConfigDiff {
+    /// Update `config` the way the matching `Collection::update_*` method does.
+    ///
+    /// Planning validates against a copy of the config, so the interpreter runs the same code on
+    /// the state itself.
+    pub fn apply(&self, config: &mut CollectionConfigInternal) -> StorageResult<()> {
+        match self {
+            CollectionConfigDiff::Optimizers(diff) => {
+                config.optimizer_config = config.optimizer_config.update(diff);
+            }
+
+            CollectionConfigDiff::Params(diff) => {
+                config.params = config.params.update(diff);
+            }
+
+            CollectionConfigDiff::Hnsw(diff) => {
+                config.hnsw_config = config.hnsw_config.update(diff);
+            }
+
+            CollectionConfigDiff::Vectors(diff) => {
+                diff.check_vector_names(&config.params)?;
+                config.params.update_vectors_from_diff(diff)?;
+            }
+
+            CollectionConfigDiff::Quantization(diff) => {
+                config.quantization_config = match diff.clone() {
+                    QuantizationConfigDiff::Scalar(scalar) => {
+                        Some(QuantizationConfig::Scalar(scalar))
+                    }
+                    QuantizationConfigDiff::Product(product) => {
+                        Some(QuantizationConfig::Product(product))
+                    }
+                    QuantizationConfigDiff::Binary(binary) => {
+                        Some(QuantizationConfig::Binary(binary))
+                    }
+                    QuantizationConfigDiff::Turbo(turbo) => Some(QuantizationConfig::Turbo(turbo)),
+                    QuantizationConfigDiff::Disabled(_) => None,
+                };
+            }
+
+            CollectionConfigDiff::SparseVectors(diff) => {
+                diff.check_vector_names(&config.params)?;
+                config.params.update_sparse_vectors_from_other(diff)?;
+            }
+
+            CollectionConfigDiff::StrictMode(diff) => {
+                config.strict_mode_config = Some(match &config.strict_mode_config {
+                    Some(current) => current.update(diff),
+                    None => diff.clone(),
+                });
+            }
+
+            // Metadata is merged, not replaced, and a null value removes its key
+            CollectionConfigDiff::Metadata(metadata) => match &mut config.metadata {
+                Some(current) => current.merge(metadata),
+                None => config.metadata = Some(metadata.clone()),
+            },
+        }
+
+        Ok(())
+    }
+}

lib/storage/src/content_manager/consensus_state_machine/mod.rs

@@ -41,7 +41,7 @@ use collection::shards::transfer::ShardTransferMethod;
 use segment::data_types::collection_defaults::CollectionConfigDefaults;
 use segment::types::HnswConfig;
 
-pub use self::action::Action;
+pub use self::action::{Action, CollectionConfigDiff};
 pub use self::state::ClusterState;
 use super::errors::StorageResult;
 use crate::content_manager::collection_meta_ops::*;
@@ -118,12 +118,24 @@ impl ConsensusStateMachine {
                 ApplyOutcome::new(self.state.plan_create_collection(&self.context, operation))
             }
 
+            CollectionMetaOperations::UpdateCollection(operation) => {
+                // TODO:
+                //
+                // Replica changes remove a replica *and* abort its transfers and resharding,
+                // so they need `Transfer::Abort`/`Resharding::Abort` to be implemented first
+
+                if operation.has_shard_replica_changes() {
+                    ApplyOutcome::NotCovered
+                } else {
+                    ApplyOutcome::new(self.state.plan_update_collection(operation))
+                }
+            }
+
             CollectionMetaOperations::DeleteCollection(operation) => {
                 ApplyOutcome::Accepted(self.state.plan_delete_collection(operation))
             }
 
-            CollectionMetaOperations::UpdateCollection(_)
-            | CollectionMetaOperations::CreateShardKey(_)
+            CollectionMetaOperations::CreateShardKey(_)
             | CollectionMetaOperations::DropShardKey(_)
             | CollectionMetaOperations::SetShardReplicaState(_)
             | CollectionMetaOperations::TransferShard(_, _)

lib/storage/src/content_manager/consensus_state_machine/state/apply.rs

@@ -17,6 +17,18 @@ impl ClusterState {
                 self.collections.remove(collection);
             }
 
+            Action::UpdateCollectionConfig { collection, diff } => {
+                let Some(state) = self.collection_mut(collection) else {
+                    return;
+                };
+
+                // Planning validates using the same function, so it should never fail here
+                if let Err(err) = diff.apply(&mut state.config) {
+                    debug_assert!(false, "rejected config diff reached the state: {err}");
+                    log::error!("Failed to update config of {collection}: {err}");
+                }
+            }
+
             Action::AddNamedVector {
                 collection,
                 vector_name,

lib/storage/src/content_manager/consensus_state_machine/state/plan.rs

@@ -8,7 +8,7 @@ use collection::shards::shard::PeerId;
 
 use super::*;
 use crate::content_manager::collection_meta_ops::*;
-use crate::content_manager::consensus_state_machine::{Action, NodeContext};
+use crate::content_manager::consensus_state_machine::{Action, CollectionConfigDiff, NodeContext};
 
 type Actions = Vec<Action>;
 
@@ -109,6 +109,111 @@ impl ClusterState {
         actions
     }
 
+    pub fn plan_update_collection(&self, op: &UpdateCollectionOperation) -> StorageResult<Actions> {
+        // TODO:
+        //
+        // `shard_replica_changes` is unimplemented, because it depends on
+        // `Transfer::Abort`/`Resharding::Abort`.
+        //
+        // If `shard_replica_changes` is set, `plan_collection_meta` returns `NotCovered`
+        // instead of calling `plan_update_collection`.
+
+        let UpdateCollectionOperation {
+            collection_name,
+            update_collection,
+            shard_replica_changes: _,
+        } = op;
+
+        let collection = self.resolve_collection(collection_name)?;
+
+        // TODO:
+        //
+        // This is intentionally different from `TableOfContent::update_collection`.
+        //
+        // `ToC::update_collection` validates and applies the diffs one by one,
+        // and saves the config after every one of them.
+        // So if a diff in the middle of the operation is rejected, every diff
+        // before it is applied and persisted, and every diff after it never runs.
+        //
+        // `plan_update_collection` validates all diffs first, and emits one
+        // `UpdateCollectionConfig` action per diff, which the applier runs in order.
+        // So either all diffs apply, or none of them do.
+        //
+        // E.g., take an operation carrying two diffs:
+        // an `hnsw_config` one, and a `vectors` one naming a vector that does not exist.
+        //
+        // `ToC::update_collection` would save the new HNSW config, then return an error.
+        // `plan_update_collection` would return an error *before* emitting any action.
+
+        // Validate operation by updating a copy of the config,
+        // so that `plan` and `apply_action` are always in sync.
+        let mut config = self
+            .collection(&collection)
+            .expect("collection exists")
+            .config
+            .clone();
+
+        // One action per field diff, in the order `ToC::update_collection` applies them.
+        //
+        // `CollectionConfigDiff` merges a diff the same way `Collection::update_*` methods do.
+        //
+        // Every diff kind is idempotent, except `Metadata` on a collection that has none.
+        //
+        // The first apply saves the whole payload as-is, `null`s included, because there is
+        // nothing to merge it into. A replay then merges the payload into what the first apply
+        // saved, and a merge *drops* every key set to `null`.
+        //
+        // E.g., take `{"a": 1, "b": null}` on a collection without metadata.
+        //
+        // The first apply saves it whole, leaving `{"a": 1, "b": null}`.
+        // A replay merges it into itself, leaving `{"a": 1}`.
+        //
+        // `replay_may_diverge` in `tests/replay.rs` exempts it.
+
+        let UpdateCollection {
+            vectors,
+            optimizers_config,
+            params,
+            hnsw_config,
+            quantization_config,
+            sparse_vectors,
+            strict_mode_config,
+            metadata,
+        } = update_collection;
+
+        let diffs = [
+            optimizers_config
+                .clone()
+                .map(CollectionConfigDiff::Optimizers),
+            params.clone().map(CollectionConfigDiff::Params),
+            (*hnsw_config).map(CollectionConfigDiff::Hnsw),
+            vectors.clone().map(CollectionConfigDiff::Vectors),
+            quantization_config
+                .clone()
+                .map(CollectionConfigDiff::Quantization),
+            sparse_vectors
+                .clone()
+                .map(CollectionConfigDiff::SparseVectors),
+            strict_mode_config
+                .clone()
+                .map(CollectionConfigDiff::StrictMode),
+            metadata.clone().map(CollectionConfigDiff::Metadata),
+        ];
+
+        let mut planned = Actions::new();
+
+        for diff in diffs.into_iter().flatten() {
+            diff.apply(&mut config)?;
+
+            planned.push(Action::UpdateCollectionConfig {
+                collection: collection.clone(),
+                diff: Box::new(diff),
+            });
+        }
+
+        Ok(planned)
+    }
+
     pub fn plan_create_named_vector(&self, op: &CreateNamedVector) -> StorageResult<Actions> {
         let CreateNamedVector {
             collection_name,

lib/storage/src/content_manager/consensus_state_machine/tests/ops.rs

@@ -2,14 +2,22 @@
 //! and tests for cases that `proptest` is unlikely to generate or reach
 
 use std::collections::{BTreeMap, BTreeSet, HashMap};
+use std::num::NonZeroU32;
 
 use ahash::AHashMap;
 use collection::collection_state::ShardInfo;
 use collection::config::ShardingMethod;
-use collection::operations::types::PeerMetadata;
+use collection::operations::config_diff::{
+    CollectionParamsDiff, HnswConfigDiff, OptimizersConfigDiff, QuantizationConfigDiff,
+};
+use collection::operations::types::{
+    PeerMetadata, SparseVectorParams, SparseVectorsConfig, VectorParamsDiff, VectorsConfigDiff,
+};
+use collection::shards::replica_set;
 use collection::shards::replica_set::replica_set_state::ReplicaState;
 use collection::shards::shard::ShardId;
 use segment::data_types::collection_defaults::CollectionConfigDefaults;
+use segment::data_types::modifier::Modifier;
 use segment::data_types::vector_name_config::*;
 use segment::types::*;
 use serde_json::{Value, json};
@@ -211,6 +219,181 @@ fn create_collection_reject_zero_shards() {
     assert_eq!(machine.state(), &state);
 }
 
+#[test]
+fn update_collection() {
+    let mut update = empty_update();
+    update.hnsw_config = Some(hnsw_diff(8));
+
+    let mut machine = state_machine(cluster_state(Vec::new()));
+    let outcome = machine.apply(&update_collection_op(update));
+
+    let ApplyOutcome::Accepted(actions) = outcome else {
+        panic!("updating a collection should be accepted, got {outcome:?}");
+    };
+
+    assert!(matches!(
+        actions.as_slice(),
+        [Action::UpdateCollectionConfig { .. }]
+    ));
+
+    assert_eq!(collection_config(&machine).hnsw_config.m, 8);
+}
+
+#[test]
+fn update_collection_diff_order() {
+    let update = UpdateCollection {
+        vectors: Some(vectors_diff("text")),
+        optimizers_config: Some(optimizers_diff()),
+        params: Some(params_diff()),
+        hnsw_config: Some(hnsw_diff(8)),
+        quantization_config: Some(QuantizationConfigDiff::new_disabled()),
+        sparse_vectors: Some(sparse_vectors_diff("sparse")),
+        strict_mode_config: Some(strict_mode_diff(true)),
+        metadata: Some(metadata(json!({ "region": "eu" }))),
+    };
+
+    let mut machine = state_machine(cluster_state(vec![
+        ("text", dense(4, Distance::Cosine)),
+        ("sparse", sparse()),
+    ]));
+
+    let outcome = machine.apply(&update_collection_op(update));
+
+    let ApplyOutcome::Accepted(actions) = outcome else {
+        panic!("updating a collection should be accepted, got {outcome:?}");
+    };
+
+    // Same order `TableOfContent::update_collection` saves them in
+    let kinds: Vec<_> = actions.iter().map(config_diff_kind).collect();
+
+    assert_eq!(
+        kinds,
+        [
+            "optimizers",
+            "params",
+            "hnsw",
+            "vectors",
+            "quantization",
+            "sparse vectors",
+            "strict mode",
+            "metadata",
+        ],
+    );
+}
+
+#[test]
+fn update_collection_replay() {
+    let mut update = empty_update();
+    update.hnsw_config = Some(hnsw_diff(8));
+    update.strict_mode_config = Some(strict_mode_diff(true));
+    update.metadata = Some(metadata(json!({ "region": "eu" })));
+
+    let mut machine = state_machine(cluster_state(Vec::new()));
+    machine.apply(&update_collection_op(update.clone()));
+
+    let applied = machine.state().clone();
+    let outcome = machine.apply(&update_collection_op(update));
+
+    let ApplyOutcome::Accepted(_) = outcome else {
+        panic!("replay of an applied update should be accepted, got {outcome:?}");
+    };
+
+    assert_eq!(
+        machine.state(),
+        &applied,
+        "replay should not change anything"
+    );
+}
+
+#[test]
+fn update_collection_reject_vector_name() {
+    let mut update = empty_update();
+    update.hnsw_config = Some(hnsw_diff(8));
+    update.vectors = Some(vectors_diff("missing"));
+
+    update_collection_reject_whole_operation(update);
+}
+
+#[test]
+fn update_collection_reject_sparse_vector_name() {
+    let mut update = empty_update();
+    update.hnsw_config = Some(hnsw_diff(8));
+    update.sparse_vectors = Some(sparse_vectors_diff("missing"));
+
+    update_collection_reject_whole_operation(update);
+}
+
+/// A diff rejected in the middle keeps nothing, not even the diffs before it
+fn update_collection_reject_whole_operation(update: UpdateCollection) {
+    let state = cluster_state(Vec::new());
+
+    let mut machine = state_machine(state.clone());
+    let outcome = machine.apply(&update_collection_op(update));
+
+    assert!(matches!(
+        outcome,
+        ApplyOutcome::Rejected(StorageError::BadInput { .. })
+    ));
+
+    assert_eq!(machine.state(), &state);
+}
+
+#[test]
+fn update_collection_metadata_merge() {
+    let mut state = cluster_state(Vec::new());
+    state
+        .collections
+        .get_mut(COLLECTION)
+        .expect("collection exists")
+        .config
+        .metadata = Some(metadata(json!({ "region": "eu", "tier": "gold" })));
+
+    let mut update = empty_update();
+    update.metadata = Some(metadata(json!({ "region": null, "size": 2 })));
+
+    let mut machine = state_machine(state);
+    machine.apply(&update_collection_op(update));
+
+    // Merged into what is there, and a null value removes its key
+    assert_eq!(
+        collection_config(&machine).metadata,
+        Some(metadata(json!({ "tier": "gold", "size": 2 }))),
+    );
+}
+
+#[test]
+fn update_collection_metadata_null_without_metadata() {
+    let mut update = empty_update();
+    update.metadata = Some(metadata(json!({ "region": null })));
+
+    let mut machine = state_machine(cluster_state(Vec::new()));
+    machine.apply(&update_collection_op(update));
+
+    // A collection with no metadata takes the payload as it is, so the null is stored.
+    // A replay merges the payload into itself and drops the key.
+    assert_eq!(
+        collection_config(&machine).metadata,
+        Some(metadata(json!({ "region": null }))),
+    );
+}
+
+#[test]
+fn update_collection_replica_changes() {
+    let mut operation = UpdateCollectionOperation::new_empty(COLLECTION.into());
+    operation.set_shard_replica_changes(vec![replica_set::Change::Remove(0, PEER_ID)]);
+
+    let state = cluster_state(Vec::new());
+
+    let mut machine = state_machine(state.clone());
+    let outcome = machine.apply(&collection_meta_op(
+        CollectionMetaOperations::UpdateCollection(operation),
+    ));
+
+    assert!(matches!(outcome, ApplyOutcome::NotCovered));
+
+    assert_eq!(machine.state(), &state);
+}
+
 #[test]
 fn delete_collection() {
     let state = cluster_state(Vec::new());
@@ -1224,6 +1407,109 @@ fn shards(placement: Vec<Vec<PeerId>>) -> AHashMap<ShardId, ShardInfo> {
         .collect()
 }
 
+fn update_collection_op(update: UpdateCollection) -> ConsensusOperations {
+    let operation =
+        UpdateCollectionOperation::new(COLLECTION.into(), update).expect("valid operation");
+
+    collection_meta_op(CollectionMetaOperations::UpdateCollection(operation))
+}
+
+/// Update with every diff absent, for a test to fill in the ones it covers
+fn empty_update() -> UpdateCollection {
+    UpdateCollectionOperation::new_empty(COLLECTION.into()).update_collection
+}
+
+fn collection_config(machine: &ConsensusStateMachine) -> &CollectionConfigInternal {
+    &machine
+        .state()
+        .collection(COLLECTION)
+        .expect("collection exists")
+        .config
+}
+
+fn config_diff_kind(action: &Action) -> &'static str {
+    let Action::UpdateCollectionConfig { diff, .. } = action else {
+        panic!("expected a config update, got {action:?}");
+    };
+
+    match **diff {
+        CollectionConfigDiff::Optimizers(_) => "optimizers",
+        CollectionConfigDiff::Params(_) => "params",
+        CollectionConfigDiff::Hnsw(_) => "hnsw",
+        CollectionConfigDiff::Vectors(_) => "vectors",
+        CollectionConfigDiff::Quantization(_) => "quantization",
+        CollectionConfigDiff::SparseVectors(_) => "sparse vectors",
+        CollectionConfigDiff::StrictMode(_) => "strict mode",
+        CollectionConfigDiff::Metadata(_) => "metadata",
+    }
+}
+
+fn hnsw_diff(m: usize) -> HnswConfigDiff {
+    HnswConfigDiff {
+        m: Some(m),
+        ..Default::default()
+    }
+}
+
+fn optimizers_diff() -> OptimizersConfigDiff {
+    OptimizersConfigDiff {
+        deleted_threshold: Some(0.5),
+        vacuum_min_vector_number: None,
+        default_segment_number: None,
+        max_segment_size: None,
+        #[expect(deprecated)]
+        memmap_threshold: None,
+        indexing_threshold: None,
+        flush_interval_sec: None,
+        max_optimization_threads: None,
+        prevent_unoptimized: None,
+    }
+}
+
+fn params_diff() -> CollectionParamsDiff {
+    CollectionParamsDiff {
+        replication_factor: NonZeroU32::new(2),
+        write_consistency_factor: None,
+        read_fan_out_factor: None,
+        read_fan_out_delay_ms: None,
+        #[expect(deprecated)]
+        on_disk_payload: None,
+        payload: None,
+    }
+}
+
+fn vectors_diff(vector_name: &str) -> VectorsConfigDiff {
+    let params = VectorParamsDiff {
+        hnsw_config: Some(hnsw_diff(8)),
+        quantization_config: None,
+        #[expect(deprecated)]
+        on_disk: None,
+        memory: None,
+    };
+
+    VectorsConfigDiff(BTreeMap::from([(vector_name.into(), params)]))
+}
+
+fn sparse_vectors_diff(vector_name: &str) -> SparseVectorsConfig {
+    let params = SparseVectorParams {
+        index: None,
+        modifier: Some(Modifier::Idf),
+    };
+
+    SparseVectorsConfig(BTreeMap::from([(vector_name.into(), params)]))
+}
+
+fn strict_mode_diff(enabled: bool) -> StrictModeConfig {
+    StrictModeConfig {
+        enabled: Some(enabled),
+        ..Default::default()
+    }
+}
+
+fn metadata(value: Value) -> Payload {
+    serde_json::from_value(value).expect("valid metadata")
+}
+
 fn delete_collection_op(collection: &str) -> ConsensusOperations {
     collection_meta_op(CollectionMetaOperations::DeleteCollection(
         DeleteCollectionOperation(collection.into()),

lib/storage/src/content_manager/consensus_state_machine/tests/prop.rs

@@ -1,10 +1,13 @@
 //! Proptest generators for cluster state and consensus operations
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
 use collection::collection_state;
 use collection::config::ShardingMethod;
-use collection::operations::types::PeerMetadata;
+use collection::operations::config_diff::{HnswConfigDiff, QuantizationConfigDiff};
+use collection::operations::types::{
+    PeerMetadata, SparseVectorParams, SparseVectorsConfig, VectorParamsDiff, VectorsConfigDiff,
+};
 use collection::shards::CollectionId;
 use collection::shards::shard::ShardId;
 use proptest::prelude::*;
@@ -173,7 +176,7 @@ pub fn arb_consensus_operation(
 
     // Weighted by how many operations each arm covers, so one operation is as likely as another
     prop_oneof![
-        8 => collection_meta,
+        9 => collection_meta,
         1 => arb_update_peer_metadata(),
         1 => arb_update_cluster_metadata(),
         1 => arb_quota_config().prop_map(ConsensusOperations::SetQuotaConfig),
@@ -188,6 +191,7 @@ fn arb_collection_meta_operation(
     prop_oneof![
         Just(CollectionMetaOperations::Nop { token: 0 }),
         arb_create_collection(collection_names.clone()),
+        arb_update_collection(collection_names.clone()),
         arb_delete_collection(collection_names.clone()),
         arb_change_aliases(collection_names.clone()),
         arb_create_named_vector(collection_names.clone()),
@@ -260,6 +264,115 @@ fn arb_shard_distribution() -> impl Strategy<Value = Option<ShardDistributionPro
     proptest::option::of(distribution)
 }
 
+/// The optimizers and params diffs are always absent. Both merge into the config exactly the way
+/// the strict-mode diff does, and that one is generated.
+fn arb_update_collection(
+    collections: Vec<String>,
+) -> impl Strategy<Value = CollectionMetaOperations> {
+    let diffs = (
+        proptest::option::of(arb_vectors_diff()),
+        proptest::option::of(arb_hnsw_diff()),
+        proptest::option::of(arb_quantization_diff()),
+        proptest::option::of(arb_sparse_vectors_diff()),
+        proptest::option::of(arb_strict_mode_diff()),
+        proptest::option::of(arb_metadata_diff()),
+    );
+
+    (arb_collection_name(collections), diffs).prop_map(|(collection_name, diffs)| {
+        let (
+            vectors,
+            hnsw_config,
+            quantization_config,
+            sparse_vectors,
+            strict_mode_config,
+            metadata,
+        ) = diffs;
+
+        let update_collection = UpdateCollection {
+            vectors,
+            optimizers_config: None,
+            params: None,
+            hnsw_config,
+            quantization_config,
+            sparse_vectors,
+            strict_mode_config,
+            metadata,
+        };
+
+        let operation = UpdateCollectionOperation::new(collection_name, update_collection)
+            .expect("valid operation");
+
+        CollectionMetaOperations::UpdateCollection(operation)
+    })
+}
+
+/// Name may be missing from the collection, or name a sparse vector, both of which are rejected
+fn arb_vectors_diff() -> impl Strategy<Value = VectorsConfigDiff> {
+    (arb_vector_name(), arb_hnsw_diff()).prop_map(|(vector_name, hnsw_config)| {
+        let params = VectorParamsDiff {
+            hnsw_config: Some(hnsw_config),
+            quantization_config: None,
+            #[expect(deprecated)]
+            on_disk: None,
+            memory: None,
+        };
+
+        VectorsConfigDiff(BTreeMap::from([(vector_name, params)]))
+    })
+}
+
+/// Name may be missing from the collection, or name a dense vector, both of which are rejected
+fn arb_sparse_vectors_diff() -> impl Strategy<Value = SparseVectorsConfig> {
+    arb_vector_name().prop_map(|vector_name| {
+        let params = SparseVectorParams {
+            index: None,
+            modifier: Some(Modifier::Idf),
+        };
+
+        SparseVectorsConfig(BTreeMap::from([(vector_name, params)]))
+    })
+}
+
+fn arb_hnsw_diff() -> impl Strategy<Value = HnswConfigDiff> {
+    proptest::sample::select(vec![8_usize, 16]).prop_map(|m| HnswConfigDiff {
+        m: Some(m),
+        ..Default::default()
+    })
+}
+
+/// Disabled clears the config, any other variant replaces it
+fn arb_quantization_diff() -> impl Strategy<Value = QuantizationConfigDiff> {
+    let scalar = ScalarQuantization {
+        scalar: ScalarQuantizationConfig {
+            r#type: ScalarType::Int8,
+            quantile: None,
+            #[expect(deprecated)]
+            always_ram: None,
+            memory: None,
+        },
+    };
+
+    prop_oneof![
+        Just(QuantizationConfigDiff::new_disabled()),
+        Just(QuantizationConfigDiff::Scalar(scalar)),
+    ]
+}
+
+fn arb_strict_mode_diff() -> impl Strategy<Value = StrictModeConfig> {
+    proptest::bool::ANY.prop_map(|enabled| StrictModeConfig {
+        enabled: Some(enabled),
+        ..Default::default()
+    })
+}
+
+/// A null value removes the key it names
+fn arb_metadata_diff() -> impl Strategy<Value = Payload> {
+    let value = prop_oneof![arb_metadata_value(), Just(serde_json::Value::Null)];
+
+    (arb_metadata_key(), value)
+        .prop_map(|(key, value)| Payload(serde_json::Map::from_iter([(key, value)])))
+}
+
 fn arb_delete_collection(
     collections: Vec<String>,
 ) -> impl Strategy<Value = CollectionMetaOperations> {

lib/storage/src/content_manager/consensus_state_machine/tests/replay.rs

@@ -1,7 +1,6 @@
 //! Correctness and replay-safety properties that must hold for every consensus operation
 
 use std::collections::HashSet;
-use std::ops::Deref as _;
 
 use proptest::prelude::*;
 
@@ -129,7 +128,14 @@ fn apply(state: &ClusterState, operation: &ConsensusOperations) -> ApplyOutcome
 }
 
 /// Operations that are not fully idempotent, and may diverge on replay.
-///
+fn replay_may_diverge(operation: &ConsensusOperations) -> bool {
+    let ConsensusOperations::CollectionMeta(operation) = operation else {
+        return false;
+    };
+
+    rename_alias_may_diverge(operation) || collection_metadata_may_diverge(operation)
+}
+
 /// `RenameAlias` is not idempotent: it moves whatever the alias points at,
 /// so a second run moves whatever the first run left under that name.
 ///
@@ -156,12 +162,8 @@ fn apply(state: &ClusterState, operation: &ConsensusOperations) -> ApplyOutcome
 /// This check is an approximate heuristic, and marks some operations that never diverge
 /// as "may diverge", such as `[prod_old → prod, prod → prod_old]`,
 /// which puts every alias back where it started.
-fn replay_may_diverge(operation: &ConsensusOperations) -> bool {
-    let ConsensusOperations::CollectionMeta(operation) = operation else {
-        return false;
-    };
-
-    let CollectionMetaOperations::ChangeAliases(operation) = operation.deref() else {
+fn rename_alias_may_diverge(operation: &CollectionMetaOperations) -> bool {
+    let CollectionMetaOperations::ChangeAliases(operation) = operation else {
         return false;
     };
 
@@ -195,3 +197,18 @@ fn replay_may_diverge(operation: &ConsensusOperations) -> bool {
 
     renames_pre_existing && renamed.is_subset(&created)
 }
+
+/// Metadata is merged into the config, where a null value removes the key it names. A collection
+/// with no metadata yet takes the whole payload instead, nulls included, and a replay merges that
+/// payload into itself and drops those keys.
+fn collection_metadata_may_diverge(operation: &CollectionMetaOperations) -> bool {
+    let CollectionMetaOperations::UpdateCollection(operation) = operation else {
+        return false;
+    };
+
+    operation
+        .update_collection
+        .metadata
+        .as_ref()
+        .is_some_and(|metadata| metadata.0.values().any(serde_json::Value::is_null))
+}