fix: lift the Kafka sink record ceiling to 16 MiB

.hyperi-ci.yaml

@@ -23,6 +23,10 @@ quality:
     - tool: osv-scanner
       id: GHSA-w9wp-h8wv-79jx
       reason: "CVE-2026-48504, opentelemetry_sdk 0.31.0 via scalo: unreachable, no Baggage propagator registered in dfe-transform-vector or scalo. Not reachable per dfe-infra docs/THREAT-MODEL.md accepted risks. Blocked on metrics-exporter-opentelemetry 0.2.1 requiring opentelemetry ^0.31; tracked in hyperi-io/scalo-rs#281."
+    # unsafe-libyaml arrives via figment and serde_yaml_ng; tracked in hyperi-io/scalo-rs#341.
+    - tool: osv-scanner
+      id: RUSTSEC-2024-0449
+      reason: "unsafe-libyaml is unmaintained, not vulnerable; it parses only operator-supplied config YAML via serde_yaml and serde_yaml_ng (figment, scalo), never untrusted input; drop when scalo moves to a maintained YAML parser"
 
 # Testing
 test:

Cargo.toml

@@ -57,8 +57,14 @@ tokio-util = { version = ">=0.7.19, <0.8", features = ["rt"] }
 # which holds a push until its records are delivered.
 #
 # `transport-kafka` carries `KafkaConfig`, the security floor `Config::validate`
-# holds the generated Kafka source and sink to. Only its config is used: the
-# image ships no librdkafka, so nothing here may build an rdkafka client.
+# holds the generated Kafka source and sink to, and `MESSAGE_MAX_BYTES`, the
+# record ceiling the generated sink takes. Only its config is used: the image
+# ships no librdkafka, so nothing here may build an rdkafka client.
+#
+# `scaling` carries the `ScalingPressure` the runtime serves at
+# `/scaling/pressure` and writes into the `scaling_pressure` gauge; the
+# supervisor drives its circuit gate directly, so it is named here as well as
+# arriving through `cli-service`.
 #
 # The floor moves with `release.helm.library` in .hyperi-ci.yaml: the library
 # renders only the contract version its scalo release writes.
@@ -72,6 +78,7 @@ scalo = { version = "2.14.9", features = [
     "lifecycle",
     "logger",
     "metrics",
+    "scaling",
     "service-metrics",
     "transport-grpc",
     "transport-grpc-vector-compat",

README.md

@@ -103,6 +103,8 @@ To see the chart a release would ship, build the binary and run `hyperi-ci chart
 
 The ScaledObject scales on CPU alone, because `contract()` sets `KafkaLagTrigger::disabled()`. Consumer-group lag rises when a downstream stage breaks, and more replicas cannot fix that. A deployment adds a trigger with the library's `keda.extraTriggers` value.
 
+scalo's runtime serves the scaling pressure at `/scaling/pressure` and writes it into the `scaling_pressure` gauge. The supervisor feeds it no components and no memory reading, so it reads 0.
+
 ## Configuration
 
 See [config.example.yaml](https://github.com/hyperi-io/dfe-transform-vector/blob/main/config.example.yaml)

deny.toml

@@ -2,7 +2,10 @@
 # Run: cargo deny check
 
 [advisories]
-ignore = []
+ignore = [
+    # unsafe-libyaml arrives via figment and serde_yaml_ng; tracked in hyperi-io/scalo-rs#341.
+    { id = "RUSTSEC-2024-0449", reason = "unsafe-libyaml is unmaintained, not vulnerable; it parses only operator-supplied config YAML via serde_yaml and serde_yaml_ng (figment, scalo), never untrusted input; drop when scalo moves to a maintained YAML parser" },
+]
 
 [licenses]
 allow = [

docs/DESIGN.md

@@ -292,6 +292,10 @@ cannot fix that. A deployment adds a trigger with the library chart's
 `keda.extraTriggers` value. `spawn_circuit_gate_task` pins scalo's scaling
 pressure to 0 while the Vector subprocess is down.
 
+scalo's runtime serves the pressure at `/scaling/pressure` and writes it into the
+`scaling_pressure` gauge. With no pressure components and no memory reading fed
+to it, the value is 0 while Vector runs too.
+
 ---
 
 ## 5. Config Engine
@@ -396,7 +400,7 @@ transforms:
   dfe_size_cap:                         # drops a record over message.max.bytes
     type: filter
     inputs: ["<last_transform_label>"]  # Auto-wired
-    condition: length(encode_json(.)) <= 999872
+    condition: length(encode_json(.)) <= 16777088
 sinks:
   dfe_sink:
     type: kafka
@@ -417,8 +421,11 @@ sinks:
       enabled: true
     librdkafka_options:
       enable.idempotence: "true"
+      message.max.bytes: "16777216"     # scalo's record ceiling unless a layer sets one
 ```
 
+The size cap and `message.max.bytes` read one value: scalo's pipeline-wide record ceiling of 16 MiB, the one every other stage's producer and every topic scalo creates carry, unless the central profile or `sink.librdkafka_options` sets `message.max.bytes`. The cap keeps 128 bytes under it for the key, headers and framing. librdkafka's own default of 1,000,000 bytes would refuse a record the rest of the chain carries.
+
 **Generated observability (`99_observability.yaml`):**
 
 ```yaml

docs/LIBRDKAFKA.md

@@ -65,7 +65,8 @@ A transform-vector pod producing to Kafka, with no central config file:
 
 Result: `linger.ms=20`, `compression.type=lz4`,
 `queue.buffering.max.kbytes=262144`, `socket.nagle.disable=true`,
-`statistics.interval.ms=1000`
+`statistics.interval.ms=1000`, and `message.max.bytes=16777216` because no layer
+set it
 
 Add a central `producer.production` naming only `linger.ms=50` and the result
 becomes `linger.ms=50`, `compression.type=lz4`,
@@ -150,7 +151,7 @@ High-throughput producer. Default for all DFE services writing to Kafka.
 | `batch.size` | `1000000` (1 MiB) | Adequate for most workloads |
 | `batch.num.messages` | `10000` | Default is reasonable |
 | `queue.buffering.max.kbytes` | `1048576` (1 GiB) | Default is fine for most services. transform-vector overrides to 256 MiB due to lower pod memory. |
-| `message.max.bytes` | `1000000` (1 MiB) | DFE messages are typically <100 KB. Services needing larger messages override individually. |
+| `message.max.bytes` | `1000000` | Left out of the profile. dfe-transform-vector writes scalo's 16 MiB pipeline ceiling where no layer sets it -- see Service-Specific Overrides. |
 | `retries` | `2147483647` (max) | Default retry-forever with idempotence is correct. |
 | `delivery.timeout.ms` | `300000` (5 min) | Generous timeout for transient broker issues |
 
@@ -202,6 +203,7 @@ baseline).
 |---------|------|-------|--------|
 | `queue.buffering.max.kbytes` | producer | `262144` (256 MiB) | Pods typically have 2-4 GiB memory. Default 1 GiB producer queue is too large. |
 | `enable.auto.commit` | consumer | `true` | Vector's kafka source stores offsets on delivery and leaves the periodic flush to librdkafka's commit timer, which the baseline's `false` never arms. Vector still commits on rebalance and clean shutdown, so the baseline stalled reported lag between those points rather than losing offsets. Lag is what KEDA scales on here. |
+| `message.max.bytes` | producer | `16777216` (16 MiB), scalo's `MESSAGE_MAX_BYTES` | The record ceiling every other stage's producer and every topic scalo creates carry; librdkafka's own `1000000` refuses a larger record. Written only where neither layer 1 nor `sink.librdkafka_options` sets the key, so either of those still decides it. The size cap in front of the sink reads the same value. |
 
 ### dfe-loader
 

src/config/generate.rs

@@ -20,10 +20,13 @@
 //!
 //! Production librdkafka defaults come from `scalo::kafka_config`
 //! (shared DFE baseline). Service-specific overrides and user-supplied
-//! `librdkafka_options` from config YAML are merged on top.
+//! `librdkafka_options` from config YAML are merged on top. The sink's
+//! `message.max.bytes`, where no layer sets it, is scalo's pipeline-wide record
+//! ceiling.
 
 use std::collections::{BTreeMap, HashMap};
 
+use scalo::transport::kafka::MESSAGE_MAX_BYTES;
 use serde_yaml_ng::Value;
 
 use super::kafka_defaults;
@@ -48,9 +51,8 @@ pub const KAFKA_COMPONENT_TYPE: &str = "kafka";
 /// librdkafka's broker hostname check, off when set to `none`.
 const HOSTNAME_VERIFICATION_KEY: &str = "ssl.endpoint.identification.algorithm";
 
-/// librdkafka's own `message.max.bytes` default, the ceiling when the sink's
-/// `librdkafka_options` sets none.
-const LIBRDKAFKA_MESSAGE_MAX_BYTES: u64 = 1_000_000;
+/// librdkafka's producer key for the largest record it puts on the wire.
+const MESSAGE_MAX_BYTES_KEY: &str = "message.max.bytes";
 
 /// Room left under `message.max.bytes` for the key, headers and record framing.
 const RECORD_FRAMING_BYTES: u64 = 128;
@@ -247,16 +249,23 @@ pub fn generate_sink_yaml(
 ///
 /// librdkafka refuses a record over `message.max.bytes`, and Vector drops a
 /// refused record. The key and headers count against the same limit, so a
-/// margin is kept below it.
+/// margin is kept below it. The ceiling is read from the options the sink is
+/// generated with, so the size cap and the producer cannot disagree.
 #[must_use]
 pub fn max_record_bytes(sink: &SinkConfig) -> u64 {
-    sink.librdkafka_options
-        .get("message.max.bytes")
+    sink_librdkafka_layers(sink)
+        .get(MESSAGE_MAX_BYTES_KEY)
         .and_then(|v| v.trim().parse::<u64>().ok())
-        .unwrap_or(LIBRDKAFKA_MESSAGE_MAX_BYTES)
+        .unwrap_or_else(pipeline_message_max_bytes)
         .saturating_sub(RECORD_FRAMING_BYTES)
 }
 
+/// scalo's pipeline-wide record ceiling, which every other stage's producer
+/// and every topic scalo creates carry.
+fn pipeline_message_max_bytes() -> u64 {
+    u64::from(MESSAGE_MAX_BYTES.unsigned_abs())
+}
+
 /// The filter in front of the Kafka sink that drops a record over
 /// [`max_record_bytes`], counted in `component_discarded_events_total`.
 ///
@@ -513,17 +522,30 @@ fn build_source_librdkafka(source: &SourceConfig) -> serde_yaml_ng::Mapping {
     m
 }
 
-fn build_sink_librdkafka(sink: &SinkConfig) -> serde_yaml_ng::Mapping {
+/// The Kafka sink's librdkafka options from every layer: the baseline profile,
+/// the service overrides, then the sink's `librdkafka_options`.
+///
+/// `message.max.bytes` takes scalo's pipeline-wide ceiling where no layer sets
+/// it: librdkafka's own default of 1,000,000 bytes refuses a record every other
+/// stage carries.
+fn sink_librdkafka_layers(sink: &SinkConfig) -> HashMap<String, String> {
     let base = kafka_defaults::producer_profile("production");
     let user_overrides: HashMap<String, String> = sink
         .librdkafka_options
         .iter()
         .map(|(k, v)| (k.clone(), v.clone()))
         .collect();
-    let merged = kafka_defaults::merge_layers(&base, SERVICE_PRODUCER_OVERRIDES, &user_overrides);
+    let mut merged =
+        kafka_defaults::merge_layers(&base, SERVICE_PRODUCER_OVERRIDES, &user_overrides);
+    merged
+        .entry(MESSAGE_MAX_BYTES_KEY.to_string())
+        .or_insert_with(|| pipeline_message_max_bytes().to_string());
+    merged
+}
 
+fn build_sink_librdkafka(sink: &SinkConfig) -> serde_yaml_ng::Mapping {
     let mut m = serde_yaml_ng::Mapping::new();
-    for (k, v) in &merged {
+    for (k, v) in &sink_librdkafka_layers(sink) {
         m.insert(val(k), val(v));
     }
 
@@ -991,8 +1013,31 @@ mod tests {
         assert!(text.contains("enable.idempotence: 'false'"));
     }
 
+    /// scalo's pipeline-wide record ceiling, which every other stage's producer
+    /// and every topic scalo creates already accept.
+    fn pipeline_ceiling() -> u64 {
+        u64::from(scalo::transport::kafka::MESSAGE_MAX_BYTES.unsigned_abs())
+    }
+
+    /// The `message.max.bytes` written into the generated Kafka sink.
+    fn written_message_max_bytes(yaml: &Value) -> Option<&str> {
+        yaml.get("sinks")
+            .and_then(|s| s.get(SINK_LABEL))
+            .and_then(|s| s.get("librdkafka_options"))
+            .and_then(|o| o.get("message.max.bytes"))
+            .and_then(Value::as_str)
+    }
+
+    /// The size cap's filter condition.
+    fn size_cap_condition(yaml: &Value) -> Option<&str> {
+        yaml.get("transforms")
+            .and_then(|t| t.get(SIZE_CAP_LABEL))
+            .and_then(|c| c.get("condition"))
+            .and_then(Value::as_str)
+    }
+
     /// The Kafka sink reads from the size cap, and the cap from the wired
-    /// inputs, with the ceiling taken from `message.max.bytes`.
+    /// inputs.
     #[test]
     fn the_kafka_sink_sits_behind_the_size_cap() {
         let yaml = generate_sink_yaml(
@@ -1018,31 +1063,59 @@ mod tests {
             cap.get("inputs").and_then(Value::as_sequence),
             Some(&vec![val("last")])
         );
-        assert_eq!(
-            cap.get("condition").and_then(Value::as_str),
-            Some("length(encode_json(.)) <= 999872")
-        );
     }
 
+    /// A sink no deployment has tuned carries a 2 MiB record. librdkafka's own
+    /// `message.max.bytes` default is 1,000,000, so both the producer and the
+    /// size cap have to sit at the ceiling the rest of the chain uses.
     #[test]
-    fn the_size_cap_follows_message_max_bytes() {
-        let mut sink = SinkConfig {
-            encoding: "raw_bytes".into(),
-            ..Default::default()
-        };
-        sink.librdkafka_options
-            .insert("message.max.bytes".into(), "2000000".into());
-        assert_eq!(max_record_bytes(&sink), 1_999_872);
+    fn a_default_sink_carries_a_2_mib_record() {
+        const TWO_MIB: u64 = 2 * 1024 * 1024;
+        let sink = SinkConfig::default();
         let yaml = generate_sink_yaml(&sink, &BridgeConfig::default(), &["src".into()], true);
+
         assert_eq!(
-            yaml.get("transforms")
-                .and_then(|t| t.get(SIZE_CAP_LABEL))
-                .and_then(|c| c.get("condition"))
-                .and_then(Value::as_str),
-            Some(r#"length(string(.message) ?? "") <= 1999872"#)
+            written_message_max_bytes(&yaml),
+            Some(pipeline_ceiling().to_string().as_str()),
+            "librdkafka refuses a record over its own 1,000,000-byte default"
+        );
+        let ceiling = pipeline_ceiling() - RECORD_FRAMING_BYTES;
+        assert_eq!(max_record_bytes(&sink), ceiling);
+        assert_eq!(
+            size_cap_condition(&yaml),
+            Some(format!("length(encode_json(.)) <= {ceiling}").as_str())
+        );
+        assert!(
+            max_record_bytes(&sink) > TWO_MIB,
+            "a 2 MiB record must pass the default size cap"
         );
     }
 
+    /// A deployment's `message.max.bytes` -- a managed broker's lower limit, or
+    /// a larger one -- is the ceiling for both librdkafka and the size cap.
+    #[test]
+    fn the_size_cap_follows_message_max_bytes() {
+        for (configured, cap) in [("2000000", 1_999_872_u64), ("8388608", 8_388_480)] {
+            let mut sink = SinkConfig {
+                encoding: "raw_bytes".into(),
+                ..Default::default()
+            };
+            sink.librdkafka_options
+                .insert("message.max.bytes".into(), configured.into());
+            assert_eq!(max_record_bytes(&sink), cap, "{configured}");
+            let yaml = generate_sink_yaml(&sink, &BridgeConfig::default(), &["src".into()], true);
+            assert_eq!(
+                written_message_max_bytes(&yaml),
+                Some(configured),
+                "the deployment's ceiling reaches librdkafka as set"
+            );
+            assert_eq!(
+                size_cap_condition(&yaml),
+                Some(format!(r#"length(string(.message) ?? "") <= {cap}"#).as_str())
+            );
+        }
+    }
+
     #[test]
     fn the_direct_sink_has_no_size_cap() {
         let sink = SinkConfig {
@@ -1230,7 +1303,8 @@ mod tests {
         // Removed settings — back to librdkafka defaults
         assert!(!text.contains("batch.size"));
         assert!(!text.contains("batch.num.messages"));
-        assert!(!text.contains("message.max.bytes"));
+        // The pipeline's record ceiling, not librdkafka's 1,000,000 default.
+        assert!(text.contains(&format!("message.max.bytes: '{}'", pipeline_ceiling())));
         // Acknowledgements on sink
         assert!(text.contains("acknowledgements:"));
         // Healthcheck

templates/bus.yaml

@@ -81,14 +81,14 @@ transforms:
       .
 
   # Fixed label. The supervisor generates this step and wires `inputs` to the
-  # last transform in the chain. librdkafka refuses a record over
-  # `message.max.bytes` (1000000 by default) and Vector drops a refused record,
-  # so this drops it first, counted, with room left for the key and headers.
+  # last transform in the chain. librdkafka refuses a record over the sink's
+  # `message.max.bytes` and Vector drops a refused record, so this drops it
+  # first, counted, with room left for the key and headers.
   dfe_size_cap:
     type: filter
     inputs:
       - dfe_transform
-    condition: length(encode_json(.)) <= 999872
+    condition: length(encode_json(.)) <= 16777088
 
 sinks:
   # Fixed label. The supervisor generates this block from `sink.*`.
@@ -111,6 +111,9 @@ sinks:
       # Retries without limit on a producer that is not idempotent reorder and
       # duplicate records within a partition.
       enable.idempotence: "true"
+      # The 16 MiB record ceiling the rest of the pipeline carries. librdkafka's
+      # own default of 1000000 refuses anything larger.
+      message.max.bytes: "16777216"
     acknowledgements:
       # The source does not commit an offset until the sink has the record.
       # Without this a broker outage loses whatever was in flight.