fix(point): escape line protocol control characters

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## 0.4.0 [unreleased]
 
+### Bug Fixes
+
+1. [#59](https://github.com/InfluxCommunity/influxdb3-rust/pull/59): Reject
+   newline, carriage return, and tab characters in measurements, tags, field
+   keys, and string field values across Point and DataFrame writes instead of
+   silently changing their stored representation.
+
 ## 0.3.0 [2026-08-27]
 
 > ⚠️ This release requires Rust 1.91 or later.

src/error.rs

@@ -74,6 +74,11 @@ pub enum Error {
     #[error("configuration error: {0}")]
     Config(String),
 
+    /// Point data contains a character that cannot be represented safely in
+    /// the structured line-protocol write paths.
+    #[error("invalid point data: {0}")]
+    InvalidPointData(String),
+
     /// Required environment variable was not set
     #[error("environment variable '{0}' is not set")]
     EnvVar(String),

src/point.rs

@@ -231,7 +231,8 @@ impl Point {
 
     /// Serialise the point to InfluxDB line protocol with the given precision.
     ///
-    /// Returns an error if the point has no fields.
+    /// Returns an error if the point has no fields or contains a newline,
+    /// carriage return, or tab in a structured line-protocol value.
     pub fn to_line_protocol(&self, precision: Precision) -> Result<String, Error> {
         let mut buf = Vec::with_capacity(64);
         let mut key_scratch = Vec::new();
@@ -260,6 +261,27 @@ impl Point {
             )));
         }
 
+        validate_line_protocol_text("measurement", &self.measurement)?;
+        for (key, value) in &self.tags {
+            validate_line_protocol_text("tag key", key)?;
+            validate_line_protocol_text("tag value", value)?;
+        }
+        for (key, value) in default_tags {
+            // Point-level tags win during the merge, so an overridden default
+            // value is never serialised and must not make the point invalid.
+            if self.tags.contains_key(key) {
+                continue;
+            }
+            validate_line_protocol_text("default tag key", key)?;
+            validate_line_protocol_text("default tag value", value)?;
+        }
+        for (key, value) in &self.fields {
+            validate_line_protocol_text("field key", key)?;
+            if let FieldValue::String(value) = value {
+                validate_line_protocol_text("string field value", value)?;
+            }
+        }
+
         // Measurement
         write_escaped_measurement(buf, &self.measurement);
 
@@ -369,8 +391,8 @@ fn tag_needs_escape(b: u8) -> bool {
     matches!(b, b',' | b'=' | b' ')
 }
 
-/// Escape a measurement name (commas and spaces). Shared with the DataFrame
-/// writer so both paths use the same rules.
+/// Escape a measurement name (commas and spaces).
+/// Shared with the DataFrame writer so both paths use the same rules.
 pub(crate) fn escape_measurement(s: &str) -> Cow<'_, str> {
     escape_with(s, measurement_needs_escape)
 }
@@ -396,6 +418,26 @@ pub(crate) fn escape_string_field(s: &str) -> Cow<'_, str> {
     Cow::Owned(out)
 }
 
+/// Reject control characters that line protocol cannot represent without
+/// changing the stored value or breaking the batch into multiple lines.
+pub(crate) fn validate_line_protocol_text(context: &str, input: &str) -> Result<(), Error> {
+    if let Some(character) = input
+        .chars()
+        .find(|character| matches!(character, '\n' | '\r' | '\t'))
+    {
+        let escaped = match character {
+            '\n' => r#"\n"#,
+            '\r' => r#"\r"#,
+            '\t' => r#"\t"#,
+            _ => unreachable!(),
+        };
+        return Err(Error::InvalidPointData(format!(
+            "{context} contains unsupported control character {escaped}"
+        )));
+    }
+    Ok(())
+}
+
 fn write_escaped_measurement(buf: &mut Vec<u8>, s: &str) {
     buf.extend_from_slice(escape_measurement(s).as_bytes());
 }

src/write_dataframe.rs

@@ -10,8 +10,9 @@ use std::collections::HashSet;
 use polars::prelude::{AnyValue, Column, DataFrame, DataType, TimeUnit};
 
 use crate::point::{
-    escape_measurement, escape_string_field, escape_tag, write_escaped_tag_value, write_lp_bool,
-    write_lp_f32, write_lp_f64, write_lp_int, write_lp_string_field, write_lp_uint,
+    escape_measurement, escape_string_field, escape_tag, validate_line_protocol_text,
+    write_escaped_tag_value, write_lp_bool, write_lp_f32, write_lp_f64, write_lp_int,
+    write_lp_string_field, write_lp_uint,
 };
 use crate::{error::Error, precision::Precision};
 
@@ -217,6 +218,8 @@ fn row_access_err(e: polars::error::PolarsError) -> Error {
 /// * Null field values omit that field for the row.
 /// * Rows where **all** fields are null are dropped entirely.
 /// * A null timestamp is omitted, so the server assigns the time.
+/// * Newline, carriage return, and tab characters in structured values return
+///   [`Error::InvalidPointData`] rather than being silently changed.
 pub fn dataframe_to_line_protocol(
     df: &DataFrame,
     measurement: &str,
@@ -230,25 +233,35 @@ pub fn dataframe_to_line_protocol(
     }
 
     let meas_escaped = escape_measurement(measurement);
+    validate_line_protocol_text("measurement", measurement)?;
     let tag_set: HashSet<&str> = tags.iter().copied().collect();
 
     // Resolve columns and escape their names once, before the row loop.
     // Missing tag columns are silently skipped (unchanged behaviour).
-    let mut tag_cols: Vec<(Cow<'_, str>, TagReader<'_>)> = tags
-        .iter()
-        .filter_map(|&t| df.column(t).ok().map(|c| (escape_tag(t), tag_reader(c))))
-        .collect();
+    let mut tag_cols: Vec<(Cow<'_, str>, TagReader<'_>)> = Vec::new();
+    for &t in tags {
+        if let Ok(c) = df.column(t) {
+            validate_line_protocol_text("tag key", t)?;
+            tag_cols.push((escape_tag(t), tag_reader(c)));
+        }
+    }
 
     // All columns that are not tag columns and not the timestamp column,
     // in frame order.
-    let mut field_cols: Vec<(Cow<'_, str>, FieldReader<'_>)> = (0..df.width())
-        .filter_map(|i| df.select_at_idx(i))
-        .filter(|c| {
+    let mut field_cols: Vec<(Cow<'_, str>, FieldReader<'_>)> = Vec::new();
+    for i in 0..df.width() {
+        let Some(c) = df.select_at_idx(i) else {
+            continue;
+        };
+        let is_field = {
             let name = c.name().as_str();
             !tag_set.contains(name) && Some(name) != timestamp_column
-        })
-        .map(|c| (escape_tag(c.name().as_str()), field_reader(c)))
-        .collect();
+        };
+        if is_field {
+            validate_line_protocol_text("field key", c.name().as_str())?;
+            field_cols.push((escape_tag(c.name().as_str()), field_reader(c)));
+        }
+    }
 
     let mut ts_reader = timestamp_column
         .and_then(|t| df.column(t).ok())
@@ -264,10 +277,19 @@ pub fn dataframe_to_line_protocol(
         buf.extend_from_slice(meas_escaped.as_bytes());
 
         // Tags are emitted in the order given by the caller.
+        // Defer tag validation errors until field presence is known because
+        // rows with all-null fields are dropped without emitting their tags.
+        let mut tag_error = None;
         for (name, reader) in tag_cols.iter_mut() {
             match reader {
                 TagReader::Str(it) => {
                     if let Some(v) = it.next().flatten() {
+                        if let Err(error) = validate_line_protocol_text("tag value", v) {
+                            if tag_error.is_none() {
+                                tag_error = Some(error);
+                            }
+                            continue;
+                        }
                         buf.push(b',');
                         buf.extend_from_slice(name.as_bytes());
                         buf.push(b'=');
@@ -277,6 +299,12 @@ pub fn dataframe_to_line_protocol(
                 TagReader::Fallback(col) => {
                     let val = col.get(row_idx).map_err(row_access_err)?;
                     if let Some(tv) = to_tag_value(val) {
+                        if let Err(error) = validate_line_protocol_text("tag value", &tv) {
+                            if tag_error.is_none() {
+                                tag_error = Some(error);
+                            }
+                            continue;
+                        }
                         buf.push(b',');
                         buf.extend_from_slice(name.as_bytes());
                         buf.push(b'=');
@@ -323,13 +351,15 @@ pub fn dataframe_to_line_protocol(
                 }
                 FieldReader::Str(it) => {
                     if let Some(v) = it.next().flatten() {
+                        validate_line_protocol_text("string field value", v)?;
                         write_field_prefix(&mut buf, &mut first, name);
                         write_lp_string_field(&mut buf, v);
                     }
                 }
                 FieldReader::Fallback(col) => {
                     let val = col.get(row_idx).map_err(row_access_err)?;
                     if let Some(fv) = to_field_value(val) {
+                        validate_line_protocol_text("field value", &fv)?;
                         write_field_prefix(&mut buf, &mut first, name);
                         buf.extend_from_slice(fv.as_bytes());
                     }
@@ -347,6 +377,10 @@ pub fn dataframe_to_line_protocol(
             continue;
         }
 
+        if let Some(error) = tag_error {
+            return Err(error);
+        }
+
         if let Some(ts) = ts {
             buf.push(b' ');
             let mut itoa_buf = itoa::Buffer::new();
@@ -678,4 +712,57 @@ mod tests {
             .unwrap();
         assert_eq!(lp, "m,host=a v=1.5 10\nm,host=b v=2.5 20");
     }
+
+    #[test]
+    fn dataframe_rejects_unsupported_control_characters() {
+        let cases = [
+            (
+                "measurement",
+                df!["v" => [1_i64]].unwrap(),
+                "me\nas",
+                &[][..],
+            ),
+            (
+                "tag key",
+                df!["tag\rkey" => ["value"], "v" => [1_i64]].unwrap(),
+                "m",
+                &["tag\rkey"][..],
+            ),
+            (
+                "field key",
+                df!["field\tkey" => [1_i64]].unwrap(),
+                "m",
+                &[][..],
+            ),
+            (
+                "string field value",
+                df!["v" => ["value\n"]].unwrap(),
+                "m",
+                &[][..],
+            ),
+        ];
+
+        for (position, df, measurement, tags) in cases {
+            let result =
+                dataframe_to_line_protocol(&df, measurement, tags, None, Precision::Nanosecond);
+            assert!(
+                matches!(result, Err(crate::error::Error::InvalidPointData(_))),
+                "{position} should be rejected, got {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn dataframe_ignores_invalid_tags_on_dropped_rows() {
+        let df = df![
+            "host" => ["invalid\ntag", "safe"],
+            "v" => [None::<i64>, Some(1_i64)],
+        ]
+        .unwrap();
+
+        let lp =
+            dataframe_to_line_protocol(&df, "m", &["host"], None, Precision::Nanosecond).unwrap();
+
+        assert_eq!(lp, "m,host=safe v=1i");
+    }
 }

tests/point_tests.rs

@@ -1,5 +1,5 @@
 /// Line-protocol serialisation tests.
-use influxdb3_client::{Point, Precision};
+use influxdb3_client::{Error, Point, Precision};
 
 #[test]
 fn full_serialisation() {
@@ -80,3 +80,42 @@ fn last_write_wins() {
     assert_eq!(lp.matches("v=").count(), 1);
     assert!(lp.contains("v=2i"));
 }
+
+#[test]
+fn line_protocol_rejects_unsupported_control_characters() {
+    let cases = [
+        ("measurement", Point::new("me\nasurement").field("v", 1_i64)),
+        (
+            "tag key",
+            Point::new("m").tag("tag\rkey", "value").field("v", 1_i64),
+        ),
+        (
+            "tag value",
+            Point::new("m").tag("key", "value\t").field("v", 1_i64),
+        ),
+        ("field key", Point::new("m").field("field\nkey", 1_i64)),
+        (
+            "string field value",
+            Point::new("m").field("field", "value\r"),
+        ),
+    ];
+
+    for (position, point) in cases {
+        let result = point.to_line_protocol(Precision::Nanosecond);
+        assert!(
+            matches!(result, Err(Error::InvalidPointData(_))),
+            "{position} should be rejected, got {result:?}"
+        );
+    }
+}
+
+#[test]
+fn line_protocol_preserves_literal_backslash_sequences() {
+    let lp = Point::new("m")
+        .tag("key", r#"literal\n"#)
+        .field("field", r#"literal\r\t"#)
+        .to_line_protocol(Precision::Nanosecond)
+        .unwrap();
+
+    assert_eq!(lp, r#"m,key=literal\n field="literal\\r\\t""#);
+}

tests/write_tests.rs

@@ -1,5 +1,5 @@
 /// Write-path integration tests against a mockito HTTP server.
-use influxdb3_client::{Client, ClientConfig, Point, Precision};
+use influxdb3_client::{Client, ClientConfig, Error, Point, Precision};
 use mockito::{Matcher, Server};
 
 async fn make_client(server: &Server) -> Client {
@@ -153,6 +153,43 @@ async fn default_tags_and_order_reach_the_wire() {
     m.assert_async().await;
 }
 
+#[tokio::test]
+async fn default_tags_reject_line_breaks_and_tabs() {
+    let server = Server::new_async().await;
+    let client = make_client(&server).await;
+
+    for (key, value) in [("env\nkey", "prod"), ("env", "prod\rvalue\t")] {
+        let result = client
+            .write(vec![Point::new("m").field("v", 1_i64)])
+            .default_tag(key, value)
+            .await;
+        assert!(
+            matches!(result, Err(Error::InvalidPointData(_))),
+            "default tag {key:?}={value:?} should be rejected, got {result:?}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn point_tag_override_ignores_invalid_default_value() {
+    let mut server = Server::new_async().await;
+    let m = server
+        .mock("POST", "/api/v3/write_lp")
+        .match_query(Matcher::Any)
+        .match_body("m,env=safe v=1i")
+        .with_status(204)
+        .create_async()
+        .await;
+
+    let client = make_client(&server).await;
+    client
+        .write(vec![Point::new("m").tag("env", "safe").field("v", 1_i64)])
+        .default_tag("env", "invalid\nvalue")
+        .await
+        .unwrap();
+    m.assert_async().await;
+}
+
 #[tokio::test]
 async fn non_retryable_error_surfaces_once() {
     // A 404 is deterministic, so it surfaces immediately without retrying.