Error logging in converters [DEVINFRA-1018]

rust/sbp/src/json/de.rs

@@ -44,12 +44,14 @@ pub fn stream_messages<R: futures::AsyncRead + Unpin>(
 #[derive(Debug, Default)]
 struct JsonDecoder {
     payload_buf: Vec<u8>,
+    eof_seen: bool,
 }
 
 impl JsonDecoder {
     fn new() -> Self {
         JsonDecoder {
             payload_buf: Vec::with_capacity(BUFLEN),
+            eof_seen: false,
         }
     }
 
@@ -75,20 +77,22 @@ impl Decoder for JsonDecoder {
     type Error = JsonError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        let value = match decode_one::<JsonInput>(src)? {
+        let value = match decode_one::<JsonInput>(&mut self.eof_seen, src)? {
             Some(v) => v,
             None => return Ok(None),
         };
-        self.parse_json(value).map(Option::Some)
+        self.parse_json(value).map(Some)
     }
 }
 
 #[derive(Debug)]
-struct Json2JsonDecoder;
+struct Json2JsonDecoder {
+    eof_seen: bool,
+}
 
 impl Json2JsonDecoder {
     fn framed<R>(input: R) -> FramedRead<R, Self> {
-        FramedRead::new(input, Self)
+        FramedRead::new(input, Self { eof_seen: false })
     }
 }
 
@@ -97,11 +101,11 @@ impl Decoder for Json2JsonDecoder {
     type Error = JsonError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        decode_one::<Json2JsonInput>(src).map_err(Into::into)
+        decode_one::<Json2JsonInput>(&mut self.eof_seen, src).map_err(Into::into)
     }
 }
 
-fn decode_one<T>(buf: &mut BytesMut) -> Result<Option<T>, serde_json::Error>
+fn decode_one<T>(eof_seen: &mut bool, buf: &mut BytesMut) -> Result<Option<T>, serde_json::Error>
 where
     T: DeserializeOwned,
 {
@@ -110,8 +114,36 @@ where
     let bytes_read = de.byte_offset();
     buf.advance(bytes_read);
     match value.transpose() {
-        Ok(v) => Ok(v),
-        Err(e) if e.is_eof() => Ok(None),
+        Ok(v) => {
+            *eof_seen = false;
+            Ok(v)
+        }
+        // One EOF error is OK, if we fail to resolve it on a second read then we return an error.
+        // (Note this will incorrectly fail if there is a JSON object that spans the underlying buffer of dencode,
+        //  which is 8k, see https://github.com/swift-nav/dencode/blob/df889f926013085212af55d0e31d59fa71f16833/src/framed_read.rs#L10)
+        Err(e) if e.is_eof() => {
+            if *eof_seen {
+                buf.advance(buf.len());
+                Err(e)
+            } else {
+                *eof_seen = true;
+                Ok(None)
+            }
+        }
+        // For semantic errors, we need to dump the invalid JSON object
+        Err(e) if e.is_data() => {
+            let mut de = Deserializer::from_slice(buf).into_iter::<serde_json::Value>();
+            let _ = de.next();
+            let bytes_read = de.byte_offset();
+            buf.advance(bytes_read);
+            Err(e)
+        }
+        // For syntax (and EOF errors) we dump the whole buffer so the stream
+        // will terminate.
+        Err(e) if e.is_syntax() => {
+            buf.advance(buf.len());
+            Err(e)
+        }
         Err(e) => Err(e),
     }
 }

rust/sbp2json/src/lib.rs

@@ -100,8 +100,12 @@ where
     E: std::error::Error + 'a,
 {
     if fatal_errors {
-        Box::new(messages.take_while(|m| m.is_ok()).map(|m| m.unwrap()))
+        Box::new(
+            messages
+                .take_while(|m| m.as_ref().map_err(|e| eprintln!("{e}")).is_ok())
+                .map(|m| m.unwrap()),
+        )
     } else {
-        Box::new(messages.filter_map(|m| m.ok()))
+        Box::new(messages.filter_map(|m| m.map_err(|e| eprintln!("{e}")).ok()))
     }
 }

rust/sbp2json/tests/common/mod.rs

@@ -257,3 +257,10 @@ fn json_file_equals(a: File, b: File) -> bool {
         true
     }
 }
+
+/// Returns relative data path
+pub fn data_path(relative_path: &str) -> PathBuf {
+    let root = find_project_root().unwrap();
+    let root = root.as_path();
+    root.join(relative_path)
+}

rust/sbp2json/tests/test_error.rs

@@ -0,0 +1,27 @@
+mod common;
+
+use crate::common::{data_path, find_project_root};
+use assert_cmd::cargo::CommandCargoExt;
+use assert_cmd::Command;
+
+/// Asserts termination of json decoder on broken inputs
+#[test]
+fn test_json_converters_termination() {
+    let input_path = data_path("test_data/errors/error_eof.json");
+    let mut cmd = Command::cargo_bin("json2sbp").unwrap();
+    cmd.arg(input_path)
+        .assert()
+        .success()
+        .stderr("expected `,` or `}` at line 1 column 185\n");
+}
+
+/// Asserts error for invalid msg_type
+#[test]
+fn test_json_untagged() {
+    let input_path = data_path("test_data/errors/error_untagged.json");
+    let mut cmd = Command::cargo_bin("json2sbp").unwrap();
+    cmd.arg(input_path)
+        .assert()
+        .success()
+        .stderr("data did not match any variant of untagged enum JsonInput\n");
+}

test_data/errors/error_eof.json

@@ -0,0 +1,2 @@
+{"crc":10555,"length":17,"msg_type":2304,"payload":"xYZjQJhhAWwAYhDy/3v+DgA=","preamble":85,"sender":51228,"tow":1080264389,"tow_f":152,"acc_x":353,"acc_y":108,"acc_z":4194,"gyr_x":114{
+{"crc":10555,"length":17,"msg_type":2304,"payload":"xYZjQJhhAWwAYhDy/3v+DgA=","preamble":85,"sender":51228,"tow":1080264389,"tow_f":152,"acc_x":353,"acc_y":108,"acc_z":4194,"gyr_x":114,"gyr_y":-389,"gyr_z":14}

test_data/errors/error_untagged.json

@@ -0,0 +1 @@
+{"crc":10555,"length":17,"msg_type":"2304","payload":"xYZjQJhhAWwAYhDy/3v+DgA=","preamble":85,"sender":51228,"tow":1080264389,"tow_f":152,"acc_x":353,"acc_y":108,"acc_z":4194,"gyr_x":114,"gyr_y":-389,"gyr_z":14}