Accept CRLF delimiters in NDJSON sync streams

powersync/src/util/line_split.rs

@@ -41,8 +41,11 @@ impl Stream for LineSplitter {
                 // Split into line including the \n, and the rest
                 let remainder = this.unfinished_line.split_off(idx + 1);
                 let mut completed_line = mem::replace(&mut this.unfinished_line, remainder);
-                // Remove \n from the completed line.
+                // Remove \n, then strip the optional \r from a CRLF delimiter.
                 completed_line.pop();
+                if completed_line.last() == Some(&b'\r') {
+                    completed_line.pop();
+                }
 
                 return Self::emit_line(completed_line);
             }
@@ -91,6 +94,26 @@ mod test {
         assert!(next.is_none());
     }
 
+    #[test]
+    fn accepts_crlf_delimiters_without_stripping_other_carriage_returns() {
+        let mut lines = LineSplitter::from(
+            stream::iter(vec![
+                Ok(Bytes::from_static(b"first\r")),
+                Ok(Bytes::from_static(b"\nsecond\r\nlast\r")),
+            ])
+            .boxed(),
+        );
+
+        let next = future::block_on(async { lines.try_next().await }).unwrap();
+        assert_eq!(next.unwrap(), "first");
+        let next = future::block_on(async { lines.try_next().await }).unwrap();
+        assert_eq!(next.unwrap(), "second");
+        let next = future::block_on(async { lines.try_next().await }).unwrap();
+        assert_eq!(next.unwrap(), "last\r");
+        let next = future::block_on(async { lines.try_next().await }).unwrap();
+        assert!(next.is_none());
+    }
+
     #[test]
     fn utf8_split_across_chunks() {
         // "é" is two bytes: 0xC3 0xA9, split across chunk boundary. This verifies we don't try to