catalog: fix review issues for cluster replica sizes

Address blocking code review feedback:

1. **Structured errors**: Replace `AdapterError::Unstructured(anyhow!(...))` with
   proper error variants:
   - `ErrorKind::ClusterReplicaSizeInUse` for in-use drop attempts
   - `ErrorKind::ReadOnlyClusterReplicaSize` for builtin drop attempts

2. **Audit log entries**: Add audit log events for CREATE and DROP
   CLUSTER REPLICA SIZE operations using `EventType::Create`/`Drop`
   with `ObjectType::ClusterReplicaSize`. Add the new variant to the
   audit log `ObjectType` enum and proto serialization.

3. **In-use safety on builtin sync**: When a size is removed from the
   env var but may still be referenced by existing replicas, mark it
   as `disabled: true` instead of deleting. This prevents panics in
   `concretize_replica_location` which expects all referenced sizes
   to exist in the catalog.

4. **Doc comment on PartialEq**: Document that the manual PartialEq
   impl on ClusterReplicaSize ignores the allocation field, and
   explain why and how to compare allocations correctly.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jubrad

src/adapter/src/catalog/open.rs

@@ -1161,10 +1161,15 @@ fn sync_builtin_cluster_replica_sizes(
         }
     }
 
-    // Remove builtins that are no longer in config.
-    for (name, _) in &durable_sizes {
+    // Remove or disable builtins that are no longer in config.
+    // We mark them disabled rather than deleting to avoid panics in
+    // concretize_replica_location if an existing replica still references the size.
+    for (name, existing) in &durable_sizes {
         if !config_sizes.0.contains_key(name) {
+            let mut disabled_allocation = existing.allocation.clone();
+            disabled_allocation.disabled = true;
             txn.remove_cluster_replica_size(name)?;
+            txn.insert_cluster_replica_size(name.clone(), disabled_allocation, true)?;
         }
     }
 

src/adapter/src/catalog/transact.rs

@@ -1716,17 +1716,30 @@ impl Catalog {
                     disabled,
                     selectors,
                 };
-                tx.insert_cluster_replica_size(name, allocation, false)?;
+                tx.insert_cluster_replica_size(name.clone(), allocation, false)?;
+
+                CatalogState::add_to_audit_log(
+                    &state.system_configuration,
+                    oracle_write_ts,
+                    session,
+                    tx,
+                    audit_events,
+                    EventType::Create,
+                    mz_audit_log::ObjectType::ClusterReplicaSize,
+                    EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                        id: name.clone(),
+                        name: name.clone(),
+                    }),
+                )?;
             }
             Op::DropClusterReplicaSize { name } => {
                 // Check that it exists and is not builtin
                 let sizes: Vec<_> = tx.get_cluster_replica_sizes().collect();
                 let size = sizes.iter().find(|s| s.name == name);
                 match size {
                     Some(s) if s.builtin => {
-                        return Err(AdapterError::Unstructured(anyhow::anyhow!(
-                            "cannot drop builtin cluster replica size '{}'",
-                            name
+                        return Err(AdapterError::Catalog(Error::new(
+                            ErrorKind::ReadOnlyClusterReplicaSize(name),
                         )));
                     }
                     Some(_) => {
@@ -1738,6 +1751,20 @@ impl Catalog {
                         )));
                     }
                 }
+
+                CatalogState::add_to_audit_log(
+                    &state.system_configuration,
+                    oracle_write_ts,
+                    session,
+                    tx,
+                    audit_events,
+                    EventType::Drop,
+                    mz_audit_log::ObjectType::ClusterReplicaSize,
+                    EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                        id: name.clone(),
+                        name: name.clone(),
+                    }),
+                )?;
             }
             Op::Comment {
                 object_id,

src/adapter/src/coord/sequencer/inner.rs

@@ -942,10 +942,13 @@ impl Coordinator {
             .state()
             .is_cluster_replica_size_in_use(&plan.name)
         {
-            return Err(AdapterError::Unstructured(anyhow::anyhow!(
-                "cannot drop cluster replica size '{}' because it is in use by an existing replica",
-                plan.name
-            )));
+            return Err(AdapterError::Catalog(
+                mz_catalog::memory::error::Error::new(
+                    mz_catalog::memory::error::ErrorKind::ClusterReplicaSizeInUse(
+                        plan.name.clone(),
+                    ),
+                ),
+            ));
         }
         let op = catalog::Op::DropClusterReplicaSize { name: plan.name };
         self.catalog_transact(None, vec![op])

src/audit-log/src/lib.rs

@@ -145,6 +145,7 @@ pub enum ObjectType {
     Func,
     Index,
     MaterializedView,
+    ClusterReplicaSize,
     NetworkPolicy,
     Role,
     Secret,
@@ -162,6 +163,7 @@ impl ObjectType {
         match self {
             ObjectType::Cluster => "Cluster",
             ObjectType::ClusterReplica => "Cluster Replica",
+            ObjectType::ClusterReplicaSize => "Cluster Replica Size",
             ObjectType::Connection => "Connection",
             ObjectType::ContinualTask => "Continual Task",
             ObjectType::Database => "Database",

src/catalog-protos/objects_hashes.json

@@ -1,7 +1,7 @@
 [
   {
     "name": "objects.rs",
-    "md5": "11000173b41468f6ff3f9593b31e5db7"
+    "md5": "c767ae77aa66bb656081a1535e9e9a62"
   },
   {
     "name": "objects_v74.rs",
@@ -37,6 +37,6 @@
   },
   {
     "name": "objects_v82.rs",
-    "md5": "11000173b41468f6ff3f9593b31e5db7"
+    "md5": "c767ae77aa66bb656081a1535e9e9a62"
   }
 ]

src/catalog-protos/src/audit_log.rs

@@ -82,6 +82,9 @@ impl RustType<crate::objects::audit_log_event_v1::ObjectType> for mz_audit_log::
             mz_audit_log::ObjectType::ClusterReplica => {
                 crate::objects::audit_log_event_v1::ObjectType::ClusterReplica
             }
+            mz_audit_log::ObjectType::ClusterReplicaSize => {
+                crate::objects::audit_log_event_v1::ObjectType::ClusterReplicaSize
+            }
             mz_audit_log::ObjectType::Connection => {
                 crate::objects::audit_log_event_v1::ObjectType::Connection
             }
@@ -133,6 +136,9 @@ impl RustType<crate::objects::audit_log_event_v1::ObjectType> for mz_audit_log::
             crate::objects::audit_log_event_v1::ObjectType::ClusterReplica => {
                 Ok(mz_audit_log::ObjectType::ClusterReplica)
             }
+            crate::objects::audit_log_event_v1::ObjectType::ClusterReplicaSize => {
+                Ok(mz_audit_log::ObjectType::ClusterReplicaSize)
+            }
             crate::objects::audit_log_event_v1::ObjectType::Connection => {
                 Ok(mz_audit_log::ObjectType::Connection)
             }

src/catalog-protos/src/objects.rs

@@ -2495,6 +2495,7 @@ pub mod audit_log_event_v1 {
         System = 16,
         ContinualTask = 17,
         NetworkPolicy = 18,
+        ClusterReplicaSize = 19,
     }
 
     #[derive(

src/catalog-protos/src/objects_v82.rs

@@ -2495,6 +2495,7 @@ pub mod audit_log_event_v1 {
         System = 16,
         ContinualTask = 17,
         NetworkPolicy = 18,
+        ClusterReplicaSize = 19,
     }
 
     #[derive(

src/catalog/src/durable/objects.rs

@@ -298,7 +298,11 @@ pub struct ClusterReplicaSize {
 }
 
 // Manual impls because ReplicaAllocation contains Numeric which doesn't impl Eq/Ord.
-// Identity is determined by name, which is the key.
+// WARNING: These impls compare only `name` and `builtin`, ignoring `allocation`.
+// Two ClusterReplicaSize values with different allocations but the same name will
+// compare equal. This is intentional for consolidation in state updates (where identity
+// is determined by key), but callers needing to detect allocation changes should
+// compare the ClusterReplicaSizeValue (via DurableType::into_key_value) instead.
 impl PartialEq for ClusterReplicaSize {
     fn eq(&self, other: &Self) -> bool {
         self.name == other.name && self.builtin == other.builtin

src/catalog/src/memory/error.rs

@@ -43,6 +43,10 @@ pub enum ErrorKind {
     ReservedNetworkPolicyName(String),
     #[error("system network policy '{0}' cannot be modified")]
     ReadOnlyNetworkPolicy(String),
+    #[error("cannot drop cluster replica size '{}' because it is in use by an existing replica", .0)]
+    ClusterReplicaSizeInUse(String),
+    #[error("cannot drop builtin cluster replica size '{}'", .0)]
+    ReadOnlyClusterReplicaSize(String),
     #[error("replica name {} is reserved", .0.quoted())]
     ReservedReplicaName(String),
     #[error("system cluster '{0}' cannot be modified")]