work on KVBuffer exhausted semantics

This commit is contained in:
Patricio Whittingslow
2026-07-29 15:37:09 -03:00
parent e346cd98d1
commit 546f11e375
9 changed files with 131 additions and 204 deletions
+3 -3
View File
@@ -1267,9 +1267,9 @@ func TestHandleBrowserSizedRequest(t *testing.T) {
sm.Reset(1)
sm.Handle("GET /echo", func(exch *Exchange) {
gotMode = string(exch.RequestHeader("X-Mode"))
exch.RequestHeaderRaw().ForEach(func(key, value []byte) error {
exch.RequestHeaderRaw().ForEach(func(key, value []byte) bool {
fields++
return nil
return true
})
})
conn := newConn(request)
@@ -1321,7 +1321,7 @@ func TestHandleRequestTooLargeAnswers431(t *testing.T) {
// Room for the fields, but not for the bytes they arrive in.
name: "more bytes than the buffer holds",
cfg: ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024, NumHeaderKVCap: 1024, NoRequestBufferGrowth: true},
wantErr: httpraw.ErrSmallHeaderBuffer,
wantErr: httpraw.ErrBufferExhausted,
},
} {
t.Run(test.name, func(t *testing.T) {
+1 -1
View File
@@ -79,7 +79,7 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error {
}
func (exch *Exchange) handleError(err error) {
if err == httpraw.ErrHeaderTooMany || err == httpraw.ErrSmallHeaderBuffer || exch.reqHdr.BufferFree() == 0 {
if err == httpraw.ErrHeaderTooMany || err == httpraw.ErrBufferExhausted || exch.reqHdr.BufferFree() == 0 {
// The peer is owed an answer: no larger buffer is coming, so
// say so instead of dropping the connection, RFC 6585 5.
exch.StageHeader("Content-Length", "0")
+1 -1
View File
@@ -55,7 +55,7 @@ func (c *Cookie) Parse() error {
break
}
if !c.kv.setInternal(k, v) {
return errOOM
return ErrBufferExhausted
}
off += n
+1 -1
View File
@@ -35,7 +35,7 @@ func (f *Form) Parse() error {
key, value, rest := NextQueryPair(f.kv.buf)
for key != nil {
if !f.kv.setInternal(key, value) {
return errOOM
return ErrBufferExhausted
}
key, value, rest = NextQueryPair(rest)
}
+13 -18
View File
@@ -234,12 +234,7 @@ func (h *Header) SetInt(key string, value int64, base int) {
if base < 2 || base > 36 {
return // strconv.AppendInt only supports base 2..36.
}
useKv := h.takeReusableSlot(key)
if useKv == nil {
h.appendHeaderInt(key, value, base)
} else {
useKv.value = h.reuseOrAppendInt(useKv.value, value, base)
}
h.hbuf.kv.SetInt(key, value, base)
}
// Set sets a key-value pair in the HTTP header.
@@ -249,9 +244,9 @@ func (h *Header) Set(key, value string) (enoughSpace bool) {
// useKv := h.takeReusableSlot(key)
// if useKv == nil {
// h.appendHeader(key, value)
// h.hbuf.kv.appendPair(key, value)
// } else {
// useKv.value = h.reuseOrAppend(useKv.value, value)
// useKv.value = h.hbuf.kv.reuseOrAppend(useKv.value, value)
// }
}
@@ -355,7 +350,7 @@ func (h *Header) ContentLength() (_ int64, present bool, _ error) {
// Add adds a new key-value pair to the HTTP header. Calling Add mangles the buffer.
func (h *Header) Add(key, value string) {
h.appendHeader(key, value)
h.hbuf.kv.appendPair(key, value)
}
// Method returns HTTP request method.
@@ -365,12 +360,12 @@ func (h *Header) Method() []byte {
// SetMethod sets the request header's method.
func (h *Header) SetMethod(method string) {
h.method = h.reuseOrAppend(h.method, method)
h.method = h.hbuf.kv.reuseOrAppend(h.method, method)
}
// SetRequestTarget sets request-target (URI) for the first HTTP request line.
func (h *Header) SetRequestTarget(requestTarget string) {
h.requestTarget = h.reuseOrAppend(h.requestTarget, requestTarget)
h.requestTarget = h.hbuf.kv.reuseOrAppend(h.requestTarget, requestTarget)
}
// RequestTarget returns a view of the request-target (URI) of the first HTTP request line.
@@ -442,7 +437,7 @@ func (h *Header) Protocol() []byte {
// SetProtocol sets the request header's protocol. Usually "HTTP/1.1".
func (h *Header) SetProtocol(protocol string) {
h.proto = h.reuseOrAppend(h.proto, protocol)
h.proto = h.hbuf.kv.reuseOrAppend(h.proto, protocol)
}
// Status returns the response header's status code and status text. i.e: "200" "OK".
@@ -456,15 +451,15 @@ func (h *Header) Status() (code, statusText []byte) {
// SetStatus sets the response header's status code and status text. i.e: "200" "OK".
func (h *Header) SetStatus(code, statusText string) {
h.hbuf.kv.flags |= FlagStatusSet
h.statusCode = h.reuseOrAppend(h.statusCode, code)
h.statusText = h.reuseOrAppend(h.statusText, statusText)
h.statusCode = h.hbuf.kv.reuseOrAppend(h.statusCode, code)
h.statusText = h.hbuf.kv.reuseOrAppend(h.statusText, statusText)
}
// SetStatusInt is identical to [Header.SetStatus] but performs integer to text conversion for status code.
func (h *Header) SetStatusInt(code int64, statusText string) {
h.hbuf.kv.flags |= FlagStatusSet
h.statusCode = h.reuseOrAppendInt(h.statusCode, code, 10)
h.statusText = h.reuseOrAppend(h.statusText, statusText)
h.statusCode = h.hbuf.kv.reuseOrAppendInt(h.statusCode, code, 10)
h.statusText = h.hbuf.kv.reuseOrAppend(h.statusText, statusText)
}
func (h *Header) getNonEmptyValue(s headerSlice) []byte {
@@ -478,7 +473,7 @@ func (h *Header) getNonEmptyValue(s headerSlice) []byte {
func (h *Header) AppendRequest(dst []byte) ([]byte, error) {
proto := h.Protocol()
if h.hbuf.kv.flags.HasAny(flagOOMReached) {
return dst, errOOM
return dst, ErrBufferExhausted
} else if h.requestTarget.len == 0 || h.method.len == 0 {
return dst, errNeedMethodURI
} else if len(proto) == 0 {
@@ -518,7 +513,7 @@ func (h *Header) AppendResponse(dst []byte) ([]byte, error) {
func (h *Header) AppendResponseNoHeaders(dst []byte) ([]byte, error) {
proto := h.Protocol()
if h.hbuf.kv.flags.HasAny(flagOOMReached) {
return dst, errOOM
return dst, ErrBufferExhausted
} else if h.statusCode.len == 0 || h.statusText.len == 0 {
return dst, errBadStatusCodeTxt
} else if len(proto) == 0 {
+2 -2
View File
@@ -473,7 +473,7 @@ func TestHeader_SplitBeforeColonStillParses(t *testing.T) {
var h Header
h.Reset(nil, numHeaderCapacity)
if _, err := h.ReadFromBytes([]byte(part1)); err != nil {
if err := h.ReadFromBytes([]byte(part1)); err != nil {
t.Fatal(err)
}
needMore, err := h.TryParse(false)
@@ -484,7 +484,7 @@ func TestHeader_SplitBeforeColonStillParses(t *testing.T) {
t.Fatal("want needMoreData=true after partial input")
}
if _, err := h.ReadFromBytes([]byte(part2)); err != nil {
if err := h.ReadFromBytes([]byte(part2)); err != nil {
t.Fatal(err)
}
needMore, err = h.TryParse(false)
+91 -39
View File
@@ -33,36 +33,34 @@ func (mb *KVBuffer) discardKVs() { mb.kvs = mb.kvs[:0] }
func (mb *KVBuffer) BufferGrowthEnabled() bool { return !mb.flags.HasAny(flagNoBufferGrow) }
func (mb *KVBuffer) ReadFromBytes(buf []byte) error {
if mb.flags.HasAny(flagMangledBuffer) {
if len(buf) == 0 {
return io.ErrNoProgress // Nothing handed over, not a buffer problem.
} else if mb.flags.HasAny(flagMangledBuffer) {
return errMangledBuffer
} else if len(buf)+cap(mb.buf) > maxBufLen {
return errOOM
return ErrBufferExhausted
}
free := mb.free()
if len(buf) > free && !mb.BufferGrowthEnabled() {
return errOOM
return ErrBufferExhausted
}
mb.buf = append(mb.buf, buf...)
return nil
}
func (mb *KVBuffer) ReadLimited(r io.Reader, limit int) (int, error) {
if mb.flags.HasAny(flagMangledBuffer) {
free := mb.free()
growthEnabled := mb.BufferGrowthEnabled()
if !growthEnabled && (free == 0 || free < limit) || len(mb.buf) >= maxBufLen {
return 0, ErrBufferExhausted
} else if mb.flags.HasAny(flagMangledBuffer) {
return 0, errMangledBuffer
} else if mb.flags.HasAny(flagReaderEOF) {
return 0, io.EOF
} else if limit <= 0 {
return 0, io.ErrNoProgress
} else if len(mb.buf) >= maxBufLen {
return 0, errOOM
}
free := mb.free()
if limit > free {
if !mb.BufferGrowthEnabled() {
return 0, errOOM
}
mb.buf = slices.Grow(mb.buf, limit)
}
mb.buf = slices.Grow(mb.buf, limit)
n, err := r.Read(mb.buf[len(mb.buf):min(len(mb.buf)+limit, maxBufLen)])
mb.buf = mb.buf[:len(mb.buf)+n]
if err != nil {
@@ -140,6 +138,37 @@ func (mb *KVBuffer) Add(key, value string) (enoughSpace bool) {
// appended with [KVBuffer.Add] and the invalidated regions are stranded, since
// nothing here compacts the buffer.
func (mb *KVBuffer) Set(key, value string) (enoughSpace bool) {
reuse := mb.takeReusableSlot(key, len(key), len(value))
if reuse < 0 {
return mb.Add(key, value)
}
mb.overwriteAt(reuse, key, value)
return true
}
// SetInt is [KVBuffer.Set]'s integer counterpart. It formats value straight into
// the slot it reuses, so overwriting a pair never allocates.
func (mb *KVBuffer) SetInt(key string, value int64, base int) (enoughSpace bool) {
reuse := mb.takeReusableSlot(key, len(key), internal.IntLen(value, base))
if reuse < 0 {
return mb.appendPairInt(key, value, base)
}
mb.flags |= flagMangledBuffer
kv := &mb.kvs[reuse]
copy(mb.buf[kv.key.start:], key)
kv.key.len = tokint(len(key))
// The slot was picked to hold keyLen/valueLen, so AppendInt writes inside
// buf and never grows a new backing array.
v := strconv.AppendInt(mb.buf[kv.value.start:kv.value.start], value, base)
kv.value.len = tokint(len(v))
return true
}
// takeReusableSlot invalidates every pair matching key except the smallest one
// whose key and value regions hold keyLen and valueLen bytes, whose index it
// returns. It returns -1 when no surviving slot fits, meaning the caller must
// append instead.
func (mb *KVBuffer) takeReusableSlot(key string, keyLen, valueLen int) int {
reuse := -1
for i := range mb.kvs {
kv := &mb.kvs[i]
@@ -147,9 +176,9 @@ func (mb *KVBuffer) Set(key, value string) (enoughSpace bool) {
continue
}
// A valueless pair holds no value region, so reusing one would write the
// value over byte 0. Let it fall through to Add, which gives the pair a
// real region and keeps "ok" distinct from "ok=".
fits := kv.HasValue() && int(kv.key.len) >= len(key) && int(kv.value.len) >= len(value)
// value over byte 0. Let it fall through to the caller's append, which
// gives the pair a real region and keeps "ok" distinct from "ok=".
fits := kv.HasValue() && int(kv.key.len) >= keyLen && int(kv.value.len) >= valueLen
if fits && (reuse < 0 || kv.size() < mb.kvs[reuse].size()) {
if reuse >= 0 {
mb.kvs[reuse].invalidate() // Superseded by a tighter fit.
@@ -159,11 +188,7 @@ func (mb *KVBuffer) Set(key, value string) (enoughSpace bool) {
}
kv.invalidate()
}
if reuse < 0 {
return mb.Add(key, value)
}
mb.overwriteAt(reuse, key, value)
return true
return reuse
}
// overwriteAt writes key and value over the regions pair i already owns. The
@@ -223,24 +248,6 @@ func (mb *KVBuffer) getIdx(key string) int {
return -1
}
func (mb *KVBuffer) getInvalidIdx() int {
for i, kv := range mb.kvs {
if !kv.isValid() {
return i
}
}
return -1
}
func (mb *KVBuffer) getInvalidOrKeyIdx(key string) int {
for i, kv := range mb.kvs {
if !kv.isValid() || key == b2s(mb.musttoken(kv.key)) {
return i
}
}
return -1
}
// reserve ensures need free bytes are available in the buffer, growing it when
// permitted. It accounts for the byte-0 reservation on an empty buffer (see
// mustAppendSlice). It returns false and sets flagOOMReached when the space
@@ -313,6 +320,51 @@ func (hb *KVBuffer) mustAppendInt(value int64, base int) headerSlice {
return hb.slice(hb.buf[L : L+len(v)])
}
// reuseOrAppend writes value over tok's slot when it fits there, avoiding any
// buffer growth; otherwise it appends a fresh slot.
func (mb *KVBuffer) reuseOrAppend(tok headerSlice, value string) headerSlice {
if tok.len > tokint(len(value)) {
copy(mb.musttoken(tok), value)
tok.len = tokint(len(value))
return tok
}
return mb.appendSlice(value)
}
// appendSlice reserves space (growing or flagging OOM) and appends value as a
// new slot.
func (mb *KVBuffer) appendSlice(value string) headerSlice {
debuglog("http:appendslice:start")
if !mb.reserve(len(value)) {
return headerSlice{} // Drop and flag OOM; never panic.
}
mb.flags |= flagMangledBuffer
return mb.mustAppendSlice(value)
}
// reuseOrAppendInt is [KVBuffer.reuseOrAppend]'s integer counterpart.
func (mb *KVBuffer) reuseOrAppendInt(tok headerSlice, value int64, base int) headerSlice {
n := internal.IntLen(value, base)
if int(tok.len) >= n {
// Reuse: format directly over the existing slot. No free space needed
// since n <= tok.len and the slot already lives inside buf.
v := strconv.AppendInt(mb.buf[tok.start:tok.start], value, base)
tok.len = tokint(len(v))
mb.flags |= flagMangledBuffer
return tok
}
return mb.appendInt(value, base, n)
}
// appendInt reserves space (growing or flagging OOM) and appends value as a new slot.
func (mb *KVBuffer) appendInt(value int64, base, n int) headerSlice {
if !mb.reserve(n) {
return headerSlice{} // Drop and flag OOM; never panic.
}
mb.flags |= flagMangledBuffer
return mb.mustAppendInt(value, base)
}
func (mb *KVBuffer) slice(value []byte) headerSlice {
if value == nil {
return headerSlice{}
+12 -126
View File
@@ -3,8 +3,6 @@ package httpraw
import (
"bytes"
"errors"
"slices"
"strconv"
"unsafe"
"github.com/soypat/lneto/internal"
@@ -18,8 +16,11 @@ var (
errNoBoundary = errors.New("httpraw: multipart boundary not set")
errUnparsed = errors.New("need to finish parsing")
errInvalidName = errors.New("invalid header name")
ErrSmallHeaderBuffer = errors.New("httpraw: Header buffer exhausted, increase size")
errOOM = errors.New("httpraw: Header incomplete due to OOM")
// ErrBufferExhausted signals a buffer with no room left for the data being
// written and no permission to grow, see [KVBuffer.EnableBufferGrowth].
// Enlarging the buffer handed to Reset is the only fix; a server answers it
// on a request header with 431, RFC 6585 5.
ErrBufferExhausted = errors.New("httpraw: buffer exhausted, increase size")
// ErrHeaderTooMany signals a header block carrying more fields than
// the buffer it is parsed into has room for, see [Header.Reset]. A server
// answers it with 431, RFC 6585 5: no larger buffer is coming, so reading
@@ -245,121 +246,6 @@ func (hb *headerBuf) parseFirstLineResponse(initFlags Flags) (statusCode, status
func (kv argsKV) HasValue() bool { return kv.value.start > 0 }
func (h *Header) reuseOrAppend(tok headerSlice, value string) headerSlice {
if tok.len > tokint(len(value)) {
copy(h.hbuf.kv.musttoken(tok), value)
tok.len = tokint(len(value))
return tok
}
return h.appendSlice(value)
}
func (h *Header) appendSlice(value string) headerSlice {
debuglog("http:appendslice:start")
if !h.reserve(len(value)) {
return headerSlice{}
}
h.flags |= flagMangledBuffer
return h.hbuf.kv.mustAppendSlice(value)
}
func (h *Header) appendHeader(key, value string) {
// reserve accounts for the byte-0 reservation mustAppendSlice makes on an
// empty buffer, and drops (flagging OOM) rather than panicking when growth
// is disabled and space runs out.
if !h.reserve(len(key) + len(value)) {
return
}
h.flags |= flagMangledBuffer
hb := &h.hbuf
k := hb.mustAppendSlice(key)
v := hb.mustAppendSlice(value)
debuglog("http:appendhdr:grow-hdrs")
hb.headers = append(hb.headers, argsKV{
key: k,
value: v,
})
}
// appendHeaderInt is appendHeader's integer counterpart: it appends key and the
// formatted integer value as a new header field.
func (h *Header) appendHeaderInt(key string, value int64, base int) {
n := internal.IntLen(value, base)
if !h.reserve(len(key) + n) {
return // Drop and flag OOM; never panic.
}
h.flags |= flagMangledBuffer
hb := &h.hbuf
k := hb.mustAppendSlice(key)
v := hb.mustAppendInt(value, base)
hb.headers = append(hb.headers, argsKV{
key: k,
value: v,
})
}
// reserve ensures need free bytes are available in the buffer, growing it when
// permitted. It accounts for the byte-0 reservation on an empty buffer (see
// mustAppendSlice). It returns false and sets flagOOMReached when the space
// cannot be guaranteed: a tokint offset overflow, or a full buffer with
// flagNoBufferGrow set.
func (h *Header) reserve(need int) bool {
hb := &h.hbuf
if len(hb.buf) == 0 {
need++ // mustAppend* reserves byte 0 on an empty buffer.
}
if len(hb.buf)+need > maxBufLen {
h.flags |= flagOOMReached // Offsets would overflow uint16 tokint.
return false
}
if need > hb.free() {
if h.flags.HasAny(flagNoBufferGrow) {
h.flags |= flagOOMReached
return false
}
hb.buf = slices.Grow(hb.buf, need)
}
return true
}
// reuseOrAppendInt writes value into tok's slot in place when it fits, avoiding
// any buffer growth; otherwise it appends a fresh slot.
func (h *Header) reuseOrAppendInt(tok headerSlice, value int64, base int) headerSlice {
n := internal.IntLen(value, base)
if int(tok.len) >= n {
// Reuse: format directly over the existing slot. No free space needed
// since n <= tok.len and the slot already lives inside buf.
v := strconv.AppendInt(h.hbuf.buf[tok.start:tok.start], value, base)
tok.len = tokint(len(v))
h.flags |= flagMangledBuffer
return tok
}
return h.appendInt(value, base, n)
}
// appendInt reserves space (growing or flagging OOM) and appends value as a new slot.
func (h *Header) appendInt(value int64, base, n int) headerSlice {
if !h.reserve(n) {
return headerSlice{} // Drop and flag OOM; never panic.
}
h.flags |= flagMangledBuffer
return h.hbuf.mustAppendInt(value, base)
}
// mustAppendInt formats value into the buffer's free region and commits it.
// The caller must have reserved at least internal.IntLen(value, base) free bytes.
func (hb *headerBuf) mustAppendInt(value int64, base int) headerSlice {
L := len(hb.buf)
if L == 0 {
L++ // Valid key-values start after byte 0.
}
v := strconv.AppendInt(hb.buf[L:L], value, base)
hb.buf = hb.buf[:L+len(v)]
return hb.slice(hb.buf[L : L+len(v)])
}
func (hb *headerBuf) noKV() argsKV { return argsKV{} }
func (hb *headerBuf) next(ss *scannerState) argsKV {
if !ss.initialized {
ss.nextColon = -1
@@ -369,10 +255,10 @@ func (hb *headerBuf) next(ss *scannerState) argsKV {
blen := len(buf)
if blen >= 2 && buf[0] == '\r' && buf[1] == '\n' {
hb.off += 2
return hb.noKV() // \r\n\r\n Ends header.
return hb.kv.noKV() // \r\n\r\n Ends header.
} else if blen >= 1 && buf[0] == '\n' {
hb.off += 1
return hb.noKV() // \n\n Ends header.
return hb.kv.noKV() // \n\n Ends header.
}
// n is parsing offset. Will start by storing colon index.
@@ -388,18 +274,18 @@ func (hb *headerBuf) next(ss *scannerState) argsKV {
// A header name should always at some point be followed by a \n
// even if it's the one that terminates the header block.
ss.err = ErrNeedMoreData
return hb.noKV()
return hb.kv.noKV()
} else if x < n {
// There was a \n before the colon! This is invalid.
ss.err = errInvalidName
return hb.noKV()
return hb.kv.noKV()
} else if n < 0 {
// A newline is present (x>=0 reached here) but the line has no
// colon: malformed, not incomplete. A split arriving before the
// colon has no newline yet and is caught by the x<0 branch above,
// so it still returns ErrNeedMoreData.
ss.err = errInvalidName
return hb.noKV()
return hb.kv.noKV()
}
}
// n stores colon position by now.
@@ -407,7 +293,7 @@ func (hb *headerBuf) next(ss *scannerState) argsKV {
// Spaces between the header key and colon are not allowed.
// See RFC 7230, Section 3.2.4.
ss.err = errInvalidName
return hb.noKV()
return hb.kv.noKV()
}
// Ready to store key..
@@ -426,7 +312,7 @@ func (hb *headerBuf) next(ss *scannerState) argsKV {
if nl < 0 || nl+n+1 == len(buf) {
// No newline or newline is last character and can't know if is multiline.
ss.err = ErrNeedMoreData
return hb.noKV()
return hb.kv.noKV()
}
n += nl + 1 // Index of the newly found newline.
nextChar := buf[n]
+7 -13
View File
@@ -17,13 +17,10 @@ func TestTryParse_IncrementalRequest(t *testing.T) {
var done bool
var doneIdx int
for i, chunk := range chunks {
n, err := hdr.ReadFromBytes([]byte(chunk))
err := hdr.ReadFromBytes([]byte(chunk))
if err != nil {
t.Fatalf("ReadFromBytes: %v", err)
}
if n != len(chunk) {
t.Fatalf("expected %d bytes read, got %d", len(chunk), n)
}
var needMore bool
needMore, err = hdr.TryParse(false)
@@ -58,13 +55,10 @@ func TestTryParse_IncrementalRequest(t *testing.T) {
// Verify headers via ForEach.
headers := make(map[string]string)
err := hdr.ForEach(func(key, value []byte) error {
hdr.ForEach(func(key, value []byte) bool {
headers[string(key)] = string(value)
return nil
return true
})
if err != nil {
t.Fatal(err)
}
if headers["Host"] != "example.com" {
t.Errorf("Host = %q; want example.com", headers["Host"])
}
@@ -176,7 +170,7 @@ func TestReadFromBytes_Empty(t *testing.T) {
var hdr Header
hdr.Reset(make([]byte, 0, 256), numHeaderCapacity)
_, err := hdr.ReadFromBytes(nil)
err := hdr.ReadFromBytes(nil)
if err == nil {
t.Fatal("expected error for empty bytes")
}
@@ -209,7 +203,7 @@ func TestEnableBufferGrowth(t *testing.T) {
for i := range big {
big[i] = 'A'
}
_, err := hdr.ReadFromBytes(big)
err := hdr.ReadFromBytes(big)
if err == nil {
t.Fatal("expected error when buffer growth disabled and data exceeds capacity")
}
@@ -228,11 +222,11 @@ func TestHeader_Add(t *testing.T) {
// ForEach should find both.
var values []string
hdr.ForEach(func(key, value []byte) error {
hdr.ForEach(func(key, value []byte) bool {
if string(key) == "X-Custom" {
values = append(values, string(value))
}
return nil
return true
})
if len(values) != 2 {
t.Fatalf("expected 2 X-Custom headers, got %d", len(values))