fix: fixed scanning after maximum segment size exceeded

Cargo.toml

@@ -4,7 +4,7 @@ members = [".", "crate/encstr"]
 [workspace.package]
 repository = "https://github.com/pamburus/hl"
 authors = ["Pavel Ivanov <mr.pavel.ivanov@gmail.com>"]
-version = "0.29.8"
+version = "0.29.9-alpha.1"
 edition = "2021"
 license = "MIT"
 
@@ -21,7 +21,7 @@ license.workspace = true
 version.workspace = true
 
 [build-dependencies]
-capnpc = "0.19"
+capnpc = "0"
 hex = "0"
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1", features = ["raw_value"] }
@@ -30,8 +30,8 @@ sha2 = "0"
 [dependencies]
 bincode = "1"
 bytefmt = "0"
-capnp = "0.19"
-chrono = { version = "0.4", default-features = false, features = ["clock", "serde", "std"] }
+capnp = "0"
+chrono = { version = "0", default-features = false, features = ["clock", "serde", "std"] }
 chrono-tz = { version = "0", features = ["serde"] }
 clap = { version = "4", features = ["wrap_help", "derive", "env", "string"] }
 clap_complete = "4"

src/app.rs

@@ -1340,6 +1340,51 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_incomplete_segment() {
+        let input = input(concat!(
+            "level=debug time=2024-01-25T19:10:20.435369+01:00 msg=m1 a.b.c=10 a.b.d=20 a.c.b=11\n",
+            "level=debug time=2024-01-25T19:10:21.764733+01:00 msg=m2 x=2\n"
+        ));
+
+        let mut output = Vec::new();
+        let app = App::new(Options {
+            buffer_size: NonZeroUsize::new(32).unwrap(),
+            max_message_size: NonZeroUsize::new(64).unwrap(),
+            ..options()
+        });
+        app.run(vec![input], &mut output).unwrap();
+        assert_eq!(
+            std::str::from_utf8(&output).unwrap(),
+            concat!(
+                "level=debug time=2024-01-25T19:10:20.435369+01:00 msg=m1 a.b.c=10 a.b.d=20 a.c.b=11\n",
+                "2024-01-25 18:10:21.764 |DBG| m2 x=2\n",
+            )
+        );
+    }
+
+    #[test]
+    fn test_incomplete_segment_sorted() {
+        let data = concat!(
+            "level=debug time=2024-01-25T19:10:20.435369+01:00 msg=m1 a.b.c=10 a.b.d=20 a.c.b=11\n",
+            "level=debug time=2024-01-25T19:10:21.764733+01:00 msg=m2 x=2\n",
+        );
+        let input = input(data);
+
+        let mut output = Vec::new();
+        let app = App::new(Options {
+            buffer_size: NonZeroUsize::new(16).unwrap(),
+            max_message_size: NonZeroUsize::new(64).unwrap(),
+            sort: true,
+            ..options()
+        });
+        app.run(vec![input], &mut output).unwrap();
+        assert_eq!(
+            std::str::from_utf8(&output).unwrap(),
+            "2024-01-25 18:10:21.764 |DBG| m2 x=2\n"
+        );
+    }
+
     #[test]
     fn test_issue_288_t1() {
         let input = input(concat!(

src/scanning.rs

@@ -812,6 +812,9 @@ impl<'a, 'b, D: Delimit> Iterator for ScannerJumboIter<'a, 'b, D> {
                         if self.can_complete() {
                             return self.complete();
                         }
+                        if placement == PartialPlacement::Last {
+                            break;
+                        }
                     }
                     next @ Some(_) => {
                         self.next = next;
@@ -824,7 +827,7 @@ impl<'a, 'b, D: Delimit> Iterator for ScannerJumboIter<'a, 'b, D> {
                         break;
                     }
                 };
-                if total > self.max_segment_size {
+                if total >= self.max_segment_size {
                     break;
                 }
             }
@@ -1045,6 +1048,28 @@ mod tests {
         )
     }
 
+    #[test]
+    fn test_jumbo_smart_new_line_2() {
+        let sf = Arc::new(SegmentBufFactory::new(3));
+        let scanner = Scanner::new(sf.clone(), SmartNewLine);
+        let mut data = std::io::Cursor::new(b"test token\r\neof\r\n");
+        let tokens = scanner
+            .items(&mut data)
+            .with_max_segment_size(9)
+            .collect::<Result<Vec<_>>>()
+            .unwrap();
+        assert_eq!(
+            tokens,
+            vec![
+                Segment::Incomplete(b"tes".into(), PartialPlacement::First),
+                Segment::Incomplete(b"t t".into(), PartialPlacement::Next),
+                Segment::Incomplete(b"oke".into(), PartialPlacement::Next),
+                Segment::Incomplete(b"n\r\n".into(), PartialPlacement::Last),
+                Segment::Complete(b"eof\r\n".into()),
+            ]
+        )
+    }
+
     #[test]
     fn test_buf_factory_recycle() {
         let factory = BufFactory::new(10);