fix: make the redis reply buffer and the http/2 request head linear

std/bytes.pith

@@ -136,6 +136,13 @@ impl ByteBuffer:
         return byte_buffer_write_byte(self.handle, value)
 
     fn write_string_utf8(text: String) -> Int!:
+        # same reasoning as write: the builtin reports bytes written and the
+        # result convention reads 0 as failure, so appending an empty string
+        # has to short-circuit rather than report an error for a no-op. text
+        # assembled piece by piece — a header value, a field that happens to
+        # be blank — is full of empty pieces.
+        if text.len() == 0:
+            return 0
         return byte_buffer_write_string_utf8(self.handle, text)
 
     fn write_line_utf8(text: String) -> Int!:
@@ -257,6 +264,13 @@ test "byte buffers write and reset":
     out.reset()
     assert_eq(out.bytes().len(), 0)
 
+    # an empty append is a no-op, not an error — the same contract write() has
+    empty_write := out.write_string_utf8("")
+    assert(not empty_write.is_err)
+    assert_eq(empty_write.ok, 0)
+    assert_eq(out.write_string_utf8("ok")!, 2)
+    assert_eq(out.bytes().to_string_utf8()!, "ok")
+
 test "byte buffers append ranges and packed words":
     src := from_string_utf8("abcdef")
     out := buffer()

std/net/http2/server.pith

@@ -638,15 +638,28 @@ fn build_request(fields: List[hpack.HeaderField], body: Bytes) -> http.HttpReque
     path := pseudo(fields, ":path")
     authority := pseudo(fields, ":authority")
     crlf := "\r\n"
-    mut head := method + " " + path + " HTTP/1.1" + crlf
-    head = head + "host: " + authority + crlf
+    # the head is appended to a byte buffer rather than to a string. pith
+    # strings are immutable, so `head = head + field` re-copies everything
+    # written so far, four times per header field. open_stream has already
+    # capped the list at MAX_HEADER_COUNT fields and MAX_HEADER_LIST_BYTES, so
+    # the worst request that reaches here is bounded — but the bound still means
+    # about twelve megabytes of copying for a head of sixty kilobytes, and this
+    # runs once per request. a buffer appends each piece once, and
+    # build_http_request wants bytes anyway.
+    head := bytes.buffer()
+    errdefer head.free()
+    head.write_string_utf8(method + " " + path + " HTTP/1.1" + crlf)!
+    head.write_string_utf8("host: " + authority + crlf)!
     for field in fields:
         # pseudo-headers are h2-only; they became the request line and Host above.
         if field.name.starts_with(":"):
             continue
-        head = head + field.name + ": " + field.value + crlf
-    head = head + crlf
-    return http.build_http_request(bytes.from_string_utf8(head), body)!
+        head.write_string_utf8(field.name)!
+        head.write_string_utf8(": ")!
+        head.write_string_utf8(field.value)!
+        head.write_string_utf8(crlf)!
+    head.write_string_utf8(crlf)!
+    return http.build_http_request(head.take_bytes(), body)!
 
 # --- the response bridge ---
 

std/redis.pith

@@ -5,9 +5,15 @@
 # a tagged value that mirrors RESP: a status line, an error, an integer, a bulk
 # string, nil, or an array of nested replies.
 #
+# a reply is text: a bulk string carrying bytes that are not utf-8 comes back
+# as an error rather than as a string quietly filled with replacement
+# characters. reach for a redis client that speaks bytes if you store blobs.
+#
 # i/o goes straight to the tcp_* builtins, the same ones std.io wraps; the
 # std.net.tcp module is a thin alias over them and isn't needed here.
 
+import std.bytes as bytes
+
 # replies come off an untrusted socket, so the length and count fields a server
 # advertises are bounded before we act on them: a hostile or buggy peer must not
 # be able to drive the client to OOM or overflow the native stack.
@@ -16,6 +22,27 @@ MAX_BULK_LEN := 536870912      # 512 MiB — a single bulk string
 MAX_ARRAY_ITEMS := 10485760    # elements in one array reply
 MAX_REPLY_DEPTH := 64          # nested-array nesting
 
+# bytes asked for in one socket read while looking for the end of a line. a
+# read allocates and zeroes everything it asks for, so asking for more than a
+# reply header needs makes every small reply pay for the one large one.
+READ_CHUNK := 4096
+
+# bytes asked for in one socket read while gathering a bulk payload of a known
+# length. the size is already known here, so a larger read is a straight saving
+# in syscalls and never over-reads: what is asked for is what is still missing,
+# up to this.
+PAYLOAD_CHUNK := 65536
+
+# the most a bulk read reserves up front. the payload length comes off the
+# wire, so reserving all of it would let a twelve-byte header ask for half a
+# gigabyte before a single payload byte has arrived; the buffer grows as the
+# data actually lands instead.
+MAX_RESERVE := 262144
+
+# the two bytes that terminate every RESP line.
+CR := 13
+LF := 10
+
 # a reply from redis. `Bulk` and `Array` may be absent, reported as `Nil`.
 # an array's elements are flattened to strings (a scalar becomes its text, a
 # nil becomes ""), which covers the common list-returning commands; reach for
@@ -44,48 +71,124 @@ fn reply_is_error(r: Reply) -> Bool:
         Reply.Error(_) => true
         _ => false
 
-# a connection to a redis server. `buf` holds bytes read past the current reply.
+# a connection to a redis server. `buf` holds the bytes read off the socket
+# that have not been parsed yet, and `pos` is how far into them the parser has
+# got. the cursor is what keeps the client linear: consuming a line or a
+# payload only moves `pos`, and the bytes already consumed are dropped in one
+# step when the next read comes in, so no byte of a reply is copied twice.
 pub struct Client:
     fd: Int
-    mut buf: String
+    mut buf: Bytes
+    mut pos: Int
 
 # open a connection to a redis server.
 pub fn connect(host: String, port: Int) -> Client!:
     fd := tcp_connect(host, port)!
-    return Client(fd: fd, buf: "")
+    return Client(fd: fd, buf: bytes.empty(), pos: 0)
 
 impl Client:
-    # pull one more chunk from the socket into the buffer; false at end of stream.
+    # bytes read off the socket that the parser has not consumed yet.
+    fn buffered() -> Int:
+        return self.buf.len() - self.pos
+
+    # decode a byte range of the buffer as text. a reply is text as far as this
+    # client is concerned, so bytes that are not utf-8 are an error rather than
+    # a string quietly filled with replacement characters.
+    fn text(start: Int, end: Int) -> String!:
+        decoded := bytes.substring_utf8(self.buf, start, end)
+        if decoded.is_err:
+            fail "redis: reply is not valid utf-8"
+        return decoded.ok
+
+    # pull one more chunk from the socket into the buffer; false at end of
+    # stream. the consumed prefix is dropped here, once per read, rather than
+    # on every parse step.
     fn fill() -> Bool!:
-        chunk := tcp_read(self.fd, 4096)!
+        chunk := tcp_read_bytes(self.fd, READ_CHUNK)!
         if chunk.len() == 0:
             return false
-        self.buf = self.buf + chunk
+        if self.buffered() == 0:
+            self.buf = chunk
+        else:
+            self.buf = self.buf.slice(self.pos, self.buf.len()).concat(chunk)
+        self.pos = 0
         return true
 
+    # index of the first CRLF at or after `start`, or -1 when there is none in
+    # the bytes read so far.
+    fn find_crlf(start: Int) -> Int:
+        mut i := start
+        if i < self.pos:
+            i = self.pos
+        limit := self.buf.len() - 1
+        while i < limit:
+            if self.buf[i] == CR and self.buf[i + 1] == LF:
+                return i
+            i = i + 1
+        return -1
+
     # read one CRLF-terminated line, returning it without the terminator.
     fn read_line() -> String!:
-        mut idx := self.buf.index_of("\r\n")
+        mut idx := self.find_crlf(self.pos)
         while idx < 0:
             # a server that never sends the terminator must not grow the buffer
             # without bound.
-            if self.buf.len() > MAX_LINE_LEN:
+            if self.buffered() > MAX_LINE_LEN:
                 fail "redis: reply line exceeds maximum length"
+            searched := self.buffered()
             if not self.fill()!:
                 fail "redis: connection closed"
-            idx = self.buf.index_of("\r\n")
-        line := self.buf.substring(0, idx)
-        self.buf = self.buf.substring(idx + 2, self.buf.len())
+            # resume one byte back, so a CR that was the last byte of the
+            # previous read still pairs with the LF that follows it. every
+            # other byte is looked at once, whatever sizes the reply arrives in.
+            mut resume := self.pos + searched - 1
+            if resume < self.pos:
+                resume = self.pos
+            idx = self.find_crlf(resume)
+        line := self.text(self.pos, idx)!
+        self.pos = idx + 2
         return line
 
     # read exactly `count` payload bytes plus the trailing CRLF.
     fn read_payload(count: Int) -> String!:
-        while self.buf.len() < count + 2:
-            if not self.fill()!:
+        need := count + 2
+        if self.buffered() >= need:
+            data := self.text(self.pos, self.pos + count)!
+            self.pos = self.pos + need
+            return data
+        # the length is known, so the rest of the payload is read straight into
+        # a buffer that grows: the payload is copied once on the way in and
+        # once on the way out, however many reads it takes to arrive.
+        mut reserve := need
+        if reserve > MAX_RESERVE:
+            reserve = MAX_RESERVE
+        out := bytes.buffer_with_capacity(reserve)
+        errdefer out.free()
+        mut got := self.buffered()
+        if got > 0:
+            # the range is the unconsumed tail by construction, so a refusal
+            # here would mean the cursor and the buffer had come apart. say so
+            # rather than carry on having silently dropped the head of a reply.
+            if not out.write_range(self.buf, self.pos, got):
+                fail "redis: buffered range is outside the read buffer"
+        self.buf = bytes.empty()
+        self.pos = 0
+        while got < need:
+            mut want := need - got
+            if want > PAYLOAD_CHUNK:
+                want = PAYLOAD_CHUNK
+            chunk := tcp_read_bytes(self.fd, want)!
+            if chunk.len() == 0:
                 fail "redis: connection closed"
-        data := self.buf.substring(0, count)
-        self.buf = self.buf.substring(count + 2, self.buf.len())
-        return data
+            out.write(chunk)!
+            got = got + chunk.len()
+        # nothing is read past the payload, so the buffer is empty afterwards
+        # and the gathered bytes are released as soon as the text is out.
+        gathered := out.take_bytes()
+        payload := bytes.substring_utf8(gathered, 0, count)
+        if payload.is_err:
+            fail "redis: reply is not valid utf-8"
+        return payload.ok
 
     # parse one reply off the wire, recursing for arrays.
     fn read_reply() -> Reply!:

tests/cases/test_http2_empty_header_value.pith

@@ -0,0 +1,99 @@
+# a header sent with an empty value must survive the http/2 request bridge.
+#
+# the bridge synthesizes an http/1.1 request head from the decoded header list,
+# and it builds that head through a byte buffer. an empty value is an empty
+# append, and an append that reports "wrote nothing" reads as a failure under
+# the result convention — so an empty value is exactly the case that turns a
+# legal request into a dropped one. this drives a real h2c request carrying
+# one empty-valued header, one ordinary header, and one whose value is empty
+# between two that are not, and checks the server saw all three.
+
+import std.net.http2.server as server
+import std.net.http2.frames as frames
+import std.net.http2.hpack as hpack
+import std.net.http as http
+import std.bytes as bytes
+import std.binary as binary
+import std.encoding as encoding
+
+# report the head the bridge synthesized, with the line breaks made visible.
+# the raw head is what proves an empty value came through: the accessor map
+# cannot tell a header with an empty value from one that was never sent.
+fn echo_headers(req: http.HttpRequestBytes) -> http.HttpResponse:
+    head := req.headers() catch "<none>"
+    return http.text(200, head.replace("\r\n", "|"))
+
+fn serve_once(listener: Int):
+    client_fd := tcp_accept(listener) catch 0
+    if client_fd != 0:
+        server.serve_h2c_connection(client_fd, echo_headers)
+
+fn send_request(fd: Int):
+    preface := encoding.from_hex("505249202a20485454502f322e300d0a0d0a534d0d0a0d0a") catch bytes.empty()
+    tcp_write_bytes(fd, preface) catch 0
+    settings := frames.frame_bytes(frames.settings_frame([]) catch frames.settings_ack_frame()) catch bytes.empty()
+    tcp_write_bytes(fd, settings) catch 0
+    encoder := hpack.new_encoder(hpack.DEFAULT_HEADER_TABLE_SIZE, false)
+    block := encoder.encode([
+        hpack.header_field(":method", "GET"),
+        hpack.header_field(":path", "/hi"),
+        hpack.header_field(":scheme", "http"),
+        hpack.header_field(":authority", "localhost"),
+        hpack.header_field("x-empty", ""),
+        hpack.header_field("x-filled", "value"),
+        hpack.header_field("x-trailing", ""),
+    ])
+    headers := frames.frame_bytes(frames.headers_frame(1, block, true, true)) catch bytes.empty()
+    tcp_write_bytes(fd, headers) catch 0
+
+# read frames until a DATA frame ends stream 1, collecting the body bytes.
+fn read_body(fd: Int) -> Bytes:
+    body := bytes.buffer()
+    mut buf := bytes.empty()
+    mut pos := 0
+    mut rounds := 0
+    while rounds < 200:
+        rounds = rounds + 1
+        while buf.len() - pos < 9:
+            chunk := tcp_read_bytes(fd, 4096) catch bytes.empty()
+            if chunk.len() == 0:
+                out := body.bytes()
+                body.free()
+                return out
+            buf = bytes.concat(buf.slice(pos, buf.len()), chunk)
+            pos = 0
+        length := binary.read_u24_be_at(buf, pos).unwrap_or(0)
+        total := 9 + length
+        while buf.len() - pos < total:
+            chunk := tcp_read_bytes(fd, 4096) catch bytes.empty()
+            if chunk.len() == 0:
+                out := body.bytes()
+                body.free()
+                return out
+            buf = bytes.concat(buf.slice(pos, buf.len()), chunk)
+            pos = 0
+        full := buf.slice(pos, pos + total)
+        pos = pos + total
+        frame := frames.read_frame(binary.reader(full), frames.MAX_ALLOWED_FRAME_SIZE) catch frames.settings_ack_frame()
+        if frame.header.frame_type == frames.FRAME_DATA:
+            df := frames.parse_data(frame) catch frames.DataFrame(0, bytes.empty(), true)
+            body.write(df.data) catch 0
+            if df.end_stream:
+                out := body.bytes()
+                body.free()
+                return out
+    out := body.bytes()
+    body.free()
+    return out
+
+fn main():
+    port := 17521
+    listener := tcp_listen("127.0.0.1", port) catch 0
+    task := spawn serve_once(listener)
+    fd := tcp_connect("127.0.0.1", port) catch 0
+    send_request(fd)
+    body := read_body(fd)
+    print(body.to_string_utf8() catch "?")
+    tcp_close(fd)
+    await task
+    tcp_close(listener)