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

core/configs/src/server_config/defaults.rs

@@ -176,6 +176,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

@@ -113,6 +113,15 @@ pub const DEFAULT_EVICTED_RING_BYTES_MAX: u64 = 16 * 1024 * 1024;
 /// trips first evicts; this byte ceiling is the second typo guard.
 pub const MAX_EVICTED_RING_BYTES: u64 = 256 * 1024 * 1024;
 
+/// Shipped default for [`PartitionConfig::dedup_clients_max`]; pinned against
+/// the runtime constant by a bootstrap assert.
+pub const PARTITION_DEDUP_CLIENTS_DEFAULT: usize = 4096;
+
+/// Ceiling for [`PartitionConfig::dedup_clients_max`]. A per-group budget, so
+/// the ceiling bounds worst-case memory at roughly `partitions * this * 146
+/// bytes`: a 112-byte slot entry plus its index-map slot.
+pub const PARTITION_DEDUP_CLIENTS_CEILING: usize = 1 << 16;
+
 /// Capacity tunables for the per-partition consensus plane.
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
 pub struct PartitionConfig {
@@ -123,6 +132,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
+    /// <= [`PARTITION_DEDUP_CLIENTS_CEILING`].
+    ///
+    /// 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 +198,14 @@ impl Validatable<ConfigurationError> for PartitionConfig {
             );
             return Err(ConfigurationError::InvalidConfigurationValue);
         }
+        if self.dedup_clients_max == 0 || self.dedup_clients_max > PARTITION_DEDUP_CLIENTS_CEILING {
+            eprintln!(
+                "{COMPONENT} partition.dedup_clients_max ({}) must be > 0 and <= \
+                 {PARTITION_DEDUP_CLIENTS_CEILING}",
+                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 +282,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn shipped_dedup_default_matches_the_runtime_constant() {
+        assert_eq!(
+            PartitionConfig::default().dedup_clients_max,
+            PARTITION_DEDUP_CLIENTS_DEFAULT,
+            "config.toml dedup_clients_max drifted from the runtime default"
+        );
+    }
+
+    #[test]
+    fn rejects_out_of_range_dedup_clients_max() {
+        for value in [0, PARTITION_DEDUP_CLIENTS_CEILING + 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 {