json-stream: preserve split CRLF delimiters across reads

When a multi-byte delimiter straddles two reads, json_stream_parse()
only scans the bytes from the latest read. As a result, a QMP reply
whose "\r\n" terminator is split between reads is not recognized.

Include up to delimiter_size - 1 trailing bytes from the existing buffer
in the next scan, so delimiters spanning reads can be matched.

Add regression coverage for a split CRLF delimiter followed by another
message in the same read. Check that both objects are parsed correctly
and that no buffered input remains afterwards.

src/libsystemd/sd-json/json-stream.c

@@ -1348,8 +1348,12 @@ int json_stream_read(JsonStream *s) {
                 }
         }
 
+        /* Also rescan the last delimiter_size - 1 bytes already buffered, so a delimiter split
+         * across reads can be found. */
+        size_t rescan = MIN(s->input_buffer_size, json_stream_delimiter_size(s) - 1);
+
         s->input_buffer_size += n;
-        s->input_buffer_unscanned += n;
+        s->input_buffer_unscanned = rescan + n;
 
         return 1;
 }

src/test/test-qmp-client.c

@@ -2,6 +2,7 @@
 
 #include <sys/eventfd.h>
 #include <sys/socket.h>
+#include <unistd.h>
 
 #include "sd-event.h"
 #include "sd-future.h"
@@ -215,6 +216,41 @@ TEST(qmp_basic) {
         run_qmp_test(mock_qmp_basic_fiber, qmp_client_basic_fiber);
 }
 
+TEST(json_stream_split_crlf_delimiter) {
+        const JsonStreamParams params = {
+                .delimiter = "\r\n",
+                .phase = mock_qmp_phase,
+                .dispatch = mock_qmp_dispatch,
+        };
+        _cleanup_(json_stream_done) JsonStream s = {};
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
+        _cleanup_close_pair_ int fds[2] = EBADF_PAIR;
+        const char part1[] = "{\"ok\":true}\r";
+        const char part2[] = "\n{\"two\":2}\r\n";
+
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, fds));
+        ASSERT_OK(json_stream_init(&s, &params));
+        int fd = TAKE_FD(fds[0]);
+        ASSERT_OK(json_stream_connect_fd_pair(&s, fd, fd));
+
+        ASSERT_OK_EQ_ERRNO(write(fds[1], part1, strlen(part1)), (ssize_t) strlen(part1));
+        ASSERT_OK_POSITIVE(json_stream_read(&s));
+        ASSERT_OK_ZERO(json_stream_parse(&s, &v));
+        ASSERT_NULL(v);
+
+        ASSERT_OK_EQ_ERRNO(write(fds[1], part2, strlen(part2)), (ssize_t) strlen(part2));
+        ASSERT_OK_POSITIVE(json_stream_read(&s));
+        ASSERT_OK_POSITIVE(json_stream_parse(&s, &v));
+        sd_json_variant *ok = ASSERT_NOT_NULL(sd_json_variant_by_key(v, "ok"));
+        ASSERT_TRUE(sd_json_variant_boolean(ok));
+
+        v = sd_json_variant_unref(v);
+        ASSERT_OK_POSITIVE(json_stream_parse(&s, &v));
+        sd_json_variant *two = ASSERT_NOT_NULL(sd_json_variant_by_key(v, "two"));
+        ASSERT_EQ(sd_json_variant_unsigned(two), 2U);
+        ASSERT_FALSE(json_stream_has_buffered_input(&s));
+}
+
 static int mock_qmp_eof_fiber(void *userdata) {
         _cleanup_(json_stream_done) JsonStream s = {};