feat(nodedb-cluster): add MetadataEntry::Batch for atomic DDL replication

Add a `Batch` variant to `MetadataEntry` that wraps a sequence of
sub-entries under a single Raft log index. The applier recurses into
batch entries via `cascade_live_state`, and the cache applies each
sub-entry in order. This ensures that a transactional DDL block
(`BEGIN; CREATE ...; COMMIT;`) either commits fully or not at all
across the cluster.

Author: farhan-syah

nodedb-cluster/src/metadata_group/applier.rs

@@ -108,6 +108,21 @@ impl CacheApplier {
         }
     }
 
+    /// Cascade live-state mutations for a committed entry. Handles
+    /// `Batch` by recursing into each sub-entry.
+    fn cascade_live_state(&self, entry: &MetadataEntry) {
+        match entry {
+            MetadataEntry::TopologyChange(change) => self.apply_topology_change(change),
+            MetadataEntry::RoutingChange(change) => self.apply_routing_change(change),
+            MetadataEntry::Batch { entries } => {
+                for sub in entries {
+                    self.cascade_live_state(sub);
+                }
+            }
+            _ => {}
+        }
+    }
+
     /// Mutate the live routing handle (if attached) in response to
     /// a committed `RoutingChange`.
     fn apply_routing_change(&self, change: &RoutingChange) {
@@ -152,12 +167,7 @@ impl MetadataApplier for CacheApplier {
             match decode_entry(data) {
                 Ok(entry) => {
                     guard.apply(*index, &entry);
-                    // Cascade to live state (if attached).
-                    match &entry {
-                        MetadataEntry::TopologyChange(change) => self.apply_topology_change(change),
-                        MetadataEntry::RoutingChange(change) => self.apply_routing_change(change),
-                        _ => {}
-                    }
+                    self.cascade_live_state(&entry);
                 }
                 Err(e) => warn!(index = *index, error = %e, "metadata decode failed"),
             }

nodedb-cluster/src/metadata_group/cache.rs

@@ -106,6 +106,11 @@ impl MetadataCache {
                 }
             }
             MetadataEntry::DescriptorDrainEnd { .. } => {}
+            MetadataEntry::Batch { entries } => {
+                for sub in entries {
+                    self.apply(index, sub);
+                }
+            }
         }
     }
 }

nodedb-cluster/src/metadata_group/entry.rs

@@ -39,6 +39,14 @@ pub enum MetadataEntry {
         payload: Vec<u8>,
     },
 
+    /// Atomic batch of metadata entries proposed by a transactional
+    /// DDL session (`BEGIN; CREATE ...; CREATE ...; COMMIT;`). The
+    /// applier unpacks and applies each sub-entry in order at a
+    /// single raft log index, so either all commit or none do.
+    Batch {
+        entries: Vec<MetadataEntry>,
+    },
+
     // ── Topology / routing ─────────────────────────────────────────────
     TopologyChange(TopologyChange),
     RoutingChange(RoutingChange),