feat(server): dedup partition writes with per-group client table slices

core/configs/src/server_config/defaults.rs

@@ -174,6 +174,7 @@ impl Default for PartitionConfig {
         let partition = &SERVER_CONFIG.partition;
         PartitionConfig {
             prepare_queue_depth: partition.prepare_queue_depth as usize,
+            dedup_clients_max: partition.dedup_clients_max as usize,
             evicted_ring_capacity: partition.evicted_ring_capacity as usize,
             evicted_ring_bytes_max: partition.evicted_ring_bytes_max.parse().unwrap(),
             transfer_served_cache_bytes_max: partition

core/configs/src/server_config/partition.rs

@@ -114,6 +114,14 @@ pub const DEFAULT_EVICTED_RING_BYTES_MAX: u64 = 16 * 1024 * 1024;
 pub const MAX_EVICTED_RING_BYTES: u64 = 256 * 1024 * 1024;
 
 /// Capacity tunables for the per-partition consensus plane.
+/// Shipped default for [`PartitionConfig::dedup_clients_max`]; pinned against
+/// the runtime constant by a bootstrap assert.
+pub const DEFAULT_PARTITION_DEDUP_CLIENTS_MAX: usize = 4096;
+
+/// Ceiling for [`PartitionConfig::dedup_clients_max`]. A per-group budget, so
+/// the ceiling bounds worst-case memory at `partitions * this * ~40 bytes`.
+pub const MAX_PARTITION_DEDUP_CLIENTS: usize = 1 << 16;
+
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
 pub struct PartitionConfig {
     /// Depth of a partition's prepare queue: how many uncommitted produce /
@@ -123,6 +131,18 @@ pub struct PartitionConfig {
     /// pinned request-buffer memory by the partition count.
     pub prepare_queue_depth: usize,
 
+    /// Distinct clients each partition group tracks request watermarks for,
+    /// deduplicating retried produces and consumer-offset writes. At capacity
+    /// the entry whose newest commit is oldest is evicted, which costs dedup
+    /// coverage for that client (its next replay re-executes, exactly as it
+    /// would have before dedup existed) and never correctness. Must be > 0 and
+    /// <= [`MAX_PARTITION_DEDUP_CLIENTS`].
+    ///
+    /// Unlike `[metadata] clients_table_max`, this budget is PER GROUP, so the
+    /// worst case scales with partition count: size it to the producers a
+    /// single partition actually sees, not the node's client total.
+    pub dedup_clients_max: usize,
+
     /// Entries the evicted ring retains per multi-replica partition for
     /// journal repair after a peer rejoins. Larger widens the window a
     /// restarting peer can be served from the ring before falling back to
@@ -177,6 +197,14 @@ impl Validatable<ConfigurationError> for PartitionConfig {
             );
             return Err(ConfigurationError::InvalidConfigurationValue);
         }
+        if self.dedup_clients_max == 0 || self.dedup_clients_max > MAX_PARTITION_DEDUP_CLIENTS {
+            eprintln!(
+                "{COMPONENT} partition.dedup_clients_max ({}) must be > 0 and <= \
+                 {MAX_PARTITION_DEDUP_CLIENTS}",
+                self.dedup_clients_max
+            );
+            return Err(ConfigurationError::InvalidConfigurationValue);
+        }
         if self.evicted_ring_capacity == 0 {
             eprintln!("{COMPONENT} partition.evicted_ring_capacity must be > 0");
             return Err(ConfigurationError::InvalidConfigurationValue);
@@ -253,6 +281,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn shipped_dedup_default_matches_the_runtime_constant() {
+        assert_eq!(
+            PartitionConfig::default().dedup_clients_max,
+            DEFAULT_PARTITION_DEDUP_CLIENTS_MAX,
+            "config.toml dedup_clients_max drifted from the runtime default"
+        );
+    }
+
+    #[test]
+    fn rejects_out_of_range_dedup_clients_max() {
+        for value in [0, MAX_PARTITION_DEDUP_CLIENTS + 1] {
+            let config = PartitionConfig {
+                dedup_clients_max: value,
+                ..PartitionConfig::default()
+            };
+            assert!(
+                config.validate().is_err(),
+                "dedup_clients_max {value} must be rejected"
+            );
+        }
+    }
+
     #[test]
     fn rejects_zero_prepare_queue_depth() {
         let config = PartitionConfig {