From b684dc54356c3f75f650e5da4b68317015012f09 Mon Sep 17 00:00:00 2001 From: rockdatasrl001 Date: Tue, 18 Aug 2026 20:39:34 +0200 Subject: [PATCH 1/8] fix(scanner): pre-size the to_the_end read buffer from a source size hint The `to_the_end` scanner reads a whole source into one message with io.ReadAll, which receives no size hint. In Go 1.26 io.ReadAll builds a list of exponentially growing chunks and then copies them into a single right-sized slice, so at peak the chunks and the final copy are live simultaneously and it holds roughly twice the content size in memory. For the `file` input this is avoidable: it already calls Stat() on every file (for the mod time) and so already knows the exact size, but never used it. This threads an optional SizeHint through SourceDetails so the `file` input can hand the scanner the size it already has, and replaces io.ReadAll in the to_the_end scanner with a capacity-hinted read that allocates the buffer once up front. The hint is an optimisation only and never a correctness input: if the file grows past the hint the buffer still grows exactly as before, and if it shrinks the read ends early, so the bytes returned are identical to io.ReadAll for any hint value. A previously latent defect is fixed as part of the plumbing: codec/scanner.go reconstructed the details object and copied only Name, silently dropping any other field. It now propagates SizeHint too (and nil-guards details). Measured on a 64 MiB in-process benchmark (Go 1.26.6, benchstat n=6): bytes allocated 157.7Mi -> 64.0Mi -59.4% (p=0.002) allocations 36 -> 2 -94.4% (p=0.002) In plain terms: to read a 64 MiB file the scanner previously reserved about 158 MiB of memory -- roughly two and a half times the file -- and did so in 36 separate allocation steps. It now reserves about 64 MiB (the file size itself) in a single step. So peak memory for the read drops from ~2.5x the file size to ~1x. For customers this is the difference that can keep large-file pipelines within a memory limit instead of being OOM-killed; the effect grows with file size (a multi-GB file previously peaked at two to three times its size in RAM). Nothing about the output or configuration changes -- it is purely a reduction in memory used. The p=0.002 figures mean the improvement is statistically solid, not measurement noise. make fmt, make lint and make test all pass. --- CHANGELOG.md | 6 + internal/component/scanner/interface.go | 6 + internal/impl/io/input_file.go | 20 +- internal/impl/io/input_file_size_test.go | 114 ++++++++++++ internal/impl/pure/scanner_to_the_end.go | 43 ++++- .../pure/scanner_to_the_end_internal_test.go | 174 ++++++++++++++++++ internal/impl/pure/scanner_to_the_end_test.go | 83 +++++++++ public/service/codec/scanner.go | 5 +- public/service/scanner.go | 14 ++ 9 files changed, 454 insertions(+), 11 deletions(-) create mode 100644 internal/impl/io/input_file_size_test.go create mode 100644 internal/impl/pure/scanner_to_the_end_internal_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f076e858e..72713e035 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ Changelog All notable changes to this project will be documented in this file. +## Unreleased + +### Fixed + +- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned. (@rockdatasrl001) + ## 4.77.0 - 2026-07-30 ### Added diff --git a/internal/component/scanner/interface.go b/internal/component/scanner/interface.go index 82c309c5b..652808aea 100644 --- a/internal/component/scanner/interface.go +++ b/internal/component/scanner/interface.go @@ -24,6 +24,12 @@ type Scanner interface { // by codec implementations in order to determine the underlying data format. type SourceDetails struct { Name string + + // SizeHint is the total number of bytes of the source, when known, and is + // zero otherwise. It is a hint only, provided so that implementations can + // pre-allocate buffers, and must never be relied upon for correctness as + // the underlying source may change between it being measured and read. + SizeHint int64 } // Creator is an interface implemented by all scanners, which allows components diff --git a/internal/impl/io/input_file.go b/internal/impl/io/input_file.go index ba906c7d1..0a1735a71 100644 --- a/internal/impl/io/input_file.go +++ b/internal/impl/io/input_file.go @@ -163,6 +163,19 @@ func (f *fileConsumer) getReader(ctx context.Context) (scannerInfo, error) { details := service.NewScannerSourceDetails() details.SetName(nextPath) + var modTimeUTC time.Time + if fInfo, err := file.Stat(); err == nil { + modTimeUTC = fInfo.ModTime().UTC() + + // Only regular files have a meaningful size to pre-allocate against, a + // FIFO, device or directory does not. + if fInfo.Mode().IsRegular() { + details.SetSizeHint(fInfo.Size()) + } + } else { + f.log.Errorf("Failed to read metadata from file '%v'", nextPath) + } + scanner, err := f.scannerCtor.Create(file, func(ctx context.Context, err error) error { if err == nil && f.delete { return f.nm.FS().Remove(nextPath) @@ -174,13 +187,6 @@ func (f *fileConsumer) getReader(ctx context.Context) (scannerInfo, error) { return scannerInfo{}, err } - var modTimeUTC time.Time - if fInfo, err := file.Stat(); err == nil { - modTimeUTC = fInfo.ModTime().UTC() - } else { - f.log.Errorf("Failed to read metadata from file '%v'", nextPath) - } - f.scannerInfo = &scannerInfo{ scanner: scanner, currentPath: nextPath, diff --git a/internal/impl/io/input_file_size_test.go b/internal/impl/io/input_file_size_test.go new file mode 100644 index 000000000..6e4aa56d9 --- /dev/null +++ b/internal/impl/io/input_file_size_test.go @@ -0,0 +1,114 @@ +// Copyright 2025 Redpanda Data, Inc. + +package io_test + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/internal/component/testutil" + "github.com/redpanda-data/benthos/v4/internal/manager/mock" + "github.com/redpanda-data/benthos/v4/internal/message" + "github.com/redpanda-data/benthos/v4/public/service" +) + +// sizeCapture records the source details observed by a scanner, so that a test +// can assert on what the file input actually propagated. +var sizeCapture = struct { + sync.Mutex + sizes map[string]int64 +}{sizes: map[string]int64{}} + +func init() { + service.MustRegisterBatchScannerCreator("capture_size_test", + service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})), + func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) { + return &captureSizeScannerCreator{}, nil + }) +} + +type captureSizeScannerCreator struct{} + +func (c *captureSizeScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { + if details != nil { + sizeCapture.Lock() + sizeCapture.sizes[filepath.Base(details.Name())] = details.SizeHint() + sizeCapture.Unlock() + } + return service.AutoAggregateBatchScannerAcks(&captureSizeScanner{r: rdr}, aFn), nil +} + +func (c *captureSizeScannerCreator) Close(context.Context) error { return nil } + +type captureSizeScanner struct { + r io.ReadCloser +} + +func (c *captureSizeScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) { + if c.r == nil { + return nil, io.EOF + } + b, err := io.ReadAll(c.r) + if err != nil { + return nil, err + } + _ = c.r.Close() + c.r = nil + return service.MessageBatch{service.NewMessage(b)}, nil +} + +func (c *captureSizeScanner) Close(ctx context.Context) error { + if c.r == nil { + return nil + } + return c.r.Close() +} + +// TestFileInputPropagatesSize asserts that the file input reads Size() from the +// Stat it already performs and passes it to the scanner, for regular files. +func TestFileInputPropagatesSize(t *testing.T) { + tmpDir := t.TempDir() + + const content = "hello world, this content has a known length" + path := filepath.Join(tmpDir, "sized.txt") + require.NoError(t, os.WriteFile(path, []byte(content), 0o644)) + + conf, err := testutil.InputFromYAML(fmt.Sprintf(` +file: + paths: [ "%v/*.txt" ] + scanner: + capture_size_test: {} +`, tmpDir)) + require.NoError(t, err) + + i, err := mock.NewManager().NewInput(conf) + require.NoError(t, err) + + i.TriggerStartConsuming() + + var tran message.Transaction + select { + case tran = <-i.TransactionChan(): + case <-time.After(time.Second): + t.Fatal("timed out") + } + + assert.Equal(t, content, string(tran.Payload.Get(0).AsBytes())) + require.NoError(t, tran.Ack(t.Context(), nil)) + + sizeCapture.Lock() + got := sizeCapture.sizes["sized.txt"] + sizeCapture.Unlock() + + assert.Equal(t, int64(len(content)), got, + "file input should propagate the size it already obtained from Stat") +} diff --git a/internal/impl/pure/scanner_to_the_end.go b/internal/impl/pure/scanner_to_the_end.go index 0b76308d3..2a92bd8d5 100644 --- a/internal/impl/pure/scanner_to_the_end.go +++ b/internal/impl/pure/scanner_to_the_end.go @@ -37,7 +37,13 @@ func toTheEndScannerCreatorFromParsed(conf *service.ParsedConfig) (s *toTheEndSc type toTheEndScannerCreator struct{} func (l *toTheEndScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { - return service.AutoAggregateBatchScannerAcks(&toTheEndScanner{r: rdr}, aFn), nil + // The size hint is used to pre-allocate the read buffer only, and is absent + // for sources of unknown length. + var sizeHint int64 + if details != nil { + sizeHint = details.SizeHint() + } + return service.AutoAggregateBatchScannerAcks(&toTheEndScanner{r: rdr, sizeHint: sizeHint}, aFn), nil } func (l *toTheEndScannerCreator) Close(context.Context) error { @@ -45,14 +51,45 @@ func (l *toTheEndScannerCreator) Close(context.Context) error { } type toTheEndScanner struct { - r io.ReadCloser + r io.ReadCloser + sizeHint int64 +} + +// readAllHinted is io.ReadAll with a starting capacity hint. +// +// Semantics are identical to io.ReadAll; the hint is purely an optimisation. +// If the source is larger than the hint the buffer grows exactly as before, +// paying the copy only for the excess; if smaller, the read ends early. This +// matters because a file can be appended to between Stat and read. +// +// The +1 on capacity is deliberate: the loop can only Read while cap > len, so +// a buffer sized exactly to the content would find len == cap on the final +// iteration and grow once more, reintroducing the reallocation this avoids. +func readAllHinted(r io.Reader, hint int64) ([]byte, error) { + if hint < 0 { + hint = 0 + } + buf := make([]byte, 0, hint+1) + for { + if len(buf) == cap(buf) { + buf = append(buf, 0)[:len(buf)] + } + n, err := r.Read(buf[len(buf):cap(buf)]) + buf = buf[:len(buf)+n] + if err != nil { + if err == io.EOF { + err = nil + } + return buf, err + } + } } func (t *toTheEndScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) { if t.r == nil { return nil, io.EOF } - mBytes, err := io.ReadAll(t.r) + mBytes, err := readAllHinted(t.r, t.sizeHint) if err != nil { return nil, err } diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go new file mode 100644 index 000000000..b0e88e02f --- /dev/null +++ b/internal/impl/pure/scanner_to_the_end_internal_test.go @@ -0,0 +1,174 @@ +// Copyright 2025 Redpanda Data, Inc. + +package pure + +import ( + "bytes" + "fmt" + "io" + "math/rand" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// shortReader wraps a reader and caps every Read at max bytes, emulating the +// short reads returned by real files and sockets. A bytes.Reader satisfies the +// whole request in a single call and so never exercises the refill path. +type shortReader struct { + r io.Reader + max int +} + +func (s *shortReader) Read(p []byte) (int, error) { + if len(p) > s.max { + p = p[:s.max] + } + return s.r.Read(p) +} + +func randomBytes(n int) []byte { + b := make([]byte, n) + // Fixed seed, so a failure reproduces exactly. + rnd := rand.New(rand.NewSource(1)) + _, _ = rnd.Read(b) + return b +} + +// TestReadAllHintedMatchesReadAll asserts that readAllHinted is a drop in +// replacement for io.ReadAll for every hint value, correct or otherwise. The +// hint is an optimisation and must never be a correctness input, as a file can +// be appended to or truncated between Stat and read. +func TestReadAllHintedMatchesReadAll(t *testing.T) { + sizes := []int{0, 1, 511, 512, 513, 4096, 1 << 20} + + for _, size := range sizes { + content := randomBytes(size) + + hints := []int64{ + 0, + 1, + int64(size / 2), + int64(size - 1), + int64(size), + int64(size + 1), + int64(size * 2), + -5, + } + + for _, hint := range hints { + for _, chunk := range []int{1, 7, 512, 1 << 16} { + t.Run(fmt.Sprintf("size=%v/hint=%v/chunk=%v", size, hint, chunk), func(t *testing.T) { + exp, err := io.ReadAll(&shortReader{r: bytes.NewReader(content), max: chunk}) + require.NoError(t, err) + + act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: chunk}, hint) + require.NoError(t, err) + + assert.Equal(t, exp, act) + assert.Len(t, act, size) + }) + } + } + } +} + +// TestReadAllHintedPropagatesErrors asserts that a non-EOF error is returned +// rather than being swallowed, matching io.ReadAll. +func TestReadAllHintedPropagatesErrors(t *testing.T) { + expErr := fmt.Errorf("nope") + + _, err := readAllHinted(io.MultiReader( + bytes.NewReader([]byte("partial")), + &errReader{err: expErr}, + ), 1024) + require.ErrorIs(t, err, expErr) +} + +type errReader struct{ err error } + +func (e *errReader) Read([]byte) (int, error) { return 0, e.err } + +// TestReadAllHintedExactHintDoesNotRealloc asserts that an accurate hint +// results in exactly one allocation, by checking the returned buffer still has +// the capacity it was created with. This is the test that catches a regression +// of the +1 subtlety: sizing the buffer to exactly the content length would +// find len == cap on the final iteration and grow once more. +func TestReadAllHintedExactHintDoesNotRealloc(t *testing.T) { + for _, size := range []int{1, 512, 4096, 1 << 20} { + t.Run(fmt.Sprintf("size=%v", size), func(t *testing.T) { + content := randomBytes(size) + + act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, int64(size)) + require.NoError(t, err) + + assert.Equal(t, content, act) + assert.Len(t, act, size) + assert.Equal(t, size+1, cap(act), "buffer was reallocated despite an exact hint") + }) + } +} + +// TestReadAllHintedGrowsWhenHintTooSmall asserts the buffer still grows to fit +// content larger than the hint, the case where a file is appended to between +// Stat and read. +func TestReadAllHintedGrowsWhenHintTooSmall(t *testing.T) { + const size = 1 << 20 + content := randomBytes(size) + + act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, 1024) + require.NoError(t, err) + assert.Equal(t, content, act) +} + +//------------------------------------------------------------------------------ + +const benchSize = 64 * 1024 * 1024 + +// repeatReader yields n bytes in fixed size short reads without allocating a +// backing buffer of its own, so the benchmark measures only the read strategy. +type repeatReader struct { + remaining int + chunk int +} + +func (c *repeatReader) Read(p []byte) (int, error) { + if c.remaining == 0 { + return 0, io.EOF + } + n := min(min(len(p), c.chunk), c.remaining) + for i := range p[:n] { + p[i] = 'x' + } + c.remaining -= n + return n, nil +} + +func BenchmarkToTheEndReadAll(b *testing.B) { + b.ReportAllocs() + b.SetBytes(benchSize) + for b.Loop() { + buf, err := io.ReadAll(&repeatReader{remaining: benchSize, chunk: 32 * 1024}) + if err != nil { + b.Fatal(err) + } + if len(buf) != benchSize { + b.Fatalf("unexpected length %v", len(buf)) + } + } +} + +func BenchmarkToTheEndReadAllHinted(b *testing.B) { + b.ReportAllocs() + b.SetBytes(benchSize) + for b.Loop() { + buf, err := readAllHinted(&repeatReader{remaining: benchSize, chunk: 32 * 1024}, benchSize) + if err != nil { + b.Fatal(err) + } + if len(buf) != benchSize { + b.Fatalf("unexpected length %v", len(buf)) + } + } +} diff --git a/internal/impl/pure/scanner_to_the_end_test.go b/internal/impl/pure/scanner_to_the_end_test.go index d94d9cbf2..d58dfcfdd 100644 --- a/internal/impl/pure/scanner_to_the_end_test.go +++ b/internal/impl/pure/scanner_to_the_end_test.go @@ -54,6 +54,89 @@ test: assert.True(t, acked) } +// TestToTheEndScannerSizeHints asserts that the content returned is unchanged +// regardless of the size hint supplied, including hints that are absent, wrong +// in either direction, or negative. The hint is an optimisation only. +func TestToTheEndScannerSizeHints(t *testing.T) { + confSpec := service.NewConfigSpec().Field(service.NewScannerField("test")) + pConf, err := confSpec.ParseYAML(` +test: + to_the_end: {} +`, nil) + require.NoError(t, err) + + rdr, err := pConf.FieldScanner("test") + require.NoError(t, err) + + const content = `firstXsecondXthird` + + for _, test := range []struct { + name string + details func() *service.ScannerSourceDetails + }{ + { + name: "no details at all", + details: func() *service.ScannerSourceDetails { return nil }, + }, + { + name: "details without a size", + details: func() *service.ScannerSourceDetails { return service.NewScannerSourceDetails() }, + }, + { + name: "exact size", + details: func() *service.ScannerSourceDetails { + d := service.NewScannerSourceDetails() + d.SetSizeHint(int64(len(content))) + return d + }, + }, + { + name: "size too small", + details: func() *service.ScannerSourceDetails { + d := service.NewScannerSourceDetails() + d.SetSizeHint(3) + return d + }, + }, + { + name: "size too large", + details: func() *service.ScannerSourceDetails { + d := service.NewScannerSourceDetails() + d.SetSizeHint(int64(len(content)) * 100) + return d + }, + }, + { + name: "negative size", + details: func() *service.ScannerSourceDetails { + d := service.NewScannerSourceDetails() + d.SetSizeHint(-5) + return d + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + buf := bytes.NewReader([]byte(content)) + strm, err := rdr.Create(io.NopCloser(buf), func(ctx context.Context, err error) error { + return nil + }, test.details()) + require.NoError(t, err) + + m, _, err := strm.NextBatch(t.Context()) + require.NoError(t, err) + require.Len(t, m, 1) + + mBytes, err := m[0].AsBytes() + require.NoError(t, err) + assert.Equal(t, content, string(mBytes)) + + _, _, err = strm.NextBatch(t.Context()) + require.Equal(t, io.EOF, err) + require.NoError(t, strm.Close(t.Context())) + }) + } +} + func TestToTheEndScannerSuite(t *testing.T) { confSpec := service.NewConfigSpec().Field(service.NewScannerField("test")) pConf, err := confSpec.ParseYAML(` diff --git a/public/service/codec/scanner.go b/public/service/codec/scanner.go index ed7487b0e..99e3bd82f 100644 --- a/public/service/codec/scanner.go +++ b/public/service/codec/scanner.go @@ -119,7 +119,10 @@ type codecRPublic struct { func (r *codecRPublic) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (DeprecatedFallbackStream, error) { sDetails := service.NewScannerSourceDetails() - sDetails.SetName(details.Name()) + if details != nil { + sDetails.SetName(details.Name()) + sDetails.SetSizeHint(details.SizeHint()) + } os, err := r.newCtor.Create(rdr, aFn, sDetails) if err != nil { diff --git a/public/service/scanner.go b/public/service/scanner.go index c512b7d61..955d78551 100644 --- a/public/service/scanner.go +++ b/public/service/scanner.go @@ -38,6 +38,20 @@ func (r *ScannerSourceDetails) Name() string { return r.details.Name } +// SetSizeHint sets the total size in bytes of the source to details. This is a +// hint only, used by scanner implementations in order to pre-allocate buffers, +// and must never be relied upon for correctness as the underlying source may +// change between it being measured and read. +func (r *ScannerSourceDetails) SetSizeHint(size int64) { + r.details.SizeHint = size +} + +// SizeHint returns the total size in bytes of the source, or zero if it has not +// been set. +func (r *ScannerSourceDetails) SizeHint() int64 { + return r.details.SizeHint +} + // BatchScannerCreator is an interface implemented by Benthos scanner plugins. // Calls to Create must create a new instantiation of BatchScanner that consumes // the provided io.ReadCloser, produces batches of messages (batches containing From 5abe58c163ff1ae5632b1d953fba56493d3de1b5 Mon Sep 17 00:00:00 2001 From: Ashley Jeffs Date: Wed, 19 Aug 2026 17:13:20 +0100 Subject: [PATCH 2/8] scanner: harden size hint handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-ups from review of the size hint work: - readAllHinted delegates to io.ReadAll when there's no usable hint — growing from an empty buffer was measurably worse than stdlib's strategy for every unhinted source (~2.4x the bytes allocated on a 64 MiB read), and unhinted sources are the common case. - Pre-allocation from a hint is capped at 1 GiB, so a wildly wrong size (procfs oddities, a custom filesystem returning garbage) can't turn into an enormous up-front allocation or a makeslice panic (hint+1 overflows at MaxInt64). Content beyond the cap still reads fine. - The decompress scanner strips the hint before handing details to its child, since it describes the compressed stream rather than the one the child actually reads. - codecRPublic passes details through wholesale instead of copying field by field, so future fields can't be silently dropped again. - ScannerSourceDetails getters are safe on a nil receiver now (details are optional everywhere), which lets the scattered nil guards go. - Docs spell out that a zero hint is indistinguishable from unset; the file input's Stat failure log includes the error value; copyright years bumped on touched files. Tests: huge-hint clamp coverage, an unhinted-path benchmark, decompress hint-strip coverage, and nil-receiver getter coverage. --- CHANGELOG.md | 2 +- internal/component/scanner/interface.go | 11 ++- internal/impl/io/input_file.go | 4 +- internal/impl/io/input_file_size_test.go | 2 +- internal/impl/pure/scanner_decompress.go | 10 +- internal/impl/pure/scanner_decompress_test.go | 98 ++++++++++++++++++- internal/impl/pure/scanner_to_the_end.go | 30 ++++-- .../pure/scanner_to_the_end_internal_test.go | 43 +++++++- internal/impl/pure/scanner_to_the_end_test.go | 2 +- public/service/codec/scanner.go | 14 ++- public/service/scanner.go | 20 +++- public/service/scanner_test.go | 19 ++++ 12 files changed, 220 insertions(+), 35 deletions(-) create mode 100644 public/service/scanner_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 72713e035..3ba5d2a5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ All notable changes to this project will be documented in this file. ### Fixed -- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned. (@rockdatasrl001) +- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned: sources of unknown size retain the exact `io.ReadAll` behaviour, pre-allocation from a hint is capped at 1 GiB to guard against wildly wrong sizes, and the `decompress` scanner strips the hint before it reaches its child, as it describes the compressed stream. (@rockdatasrl001) ## 4.77.0 - 2026-07-30 diff --git a/internal/component/scanner/interface.go b/internal/component/scanner/interface.go index 652808aea..6ee2158cc 100644 --- a/internal/component/scanner/interface.go +++ b/internal/component/scanner/interface.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package scanner @@ -26,9 +26,12 @@ type SourceDetails struct { Name string // SizeHint is the total number of bytes of the source, when known, and is - // zero otherwise. It is a hint only, provided so that implementations can - // pre-allocate buffers, and must never be relied upon for correctness as - // the underlying source may change between it being measured and read. + // zero otherwise. A source measured as empty is therefore + // indistinguishable from one of unknown size, and implementations must + // treat zero as unknown. It is a hint only, provided so that + // implementations can pre-allocate buffers, and must never be relied upon + // for correctness as the underlying source may change between it being + // measured and read. SizeHint int64 } diff --git a/internal/impl/io/input_file.go b/internal/impl/io/input_file.go index 0a1735a71..278452418 100644 --- a/internal/impl/io/input_file.go +++ b/internal/impl/io/input_file.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package io @@ -173,7 +173,7 @@ func (f *fileConsumer) getReader(ctx context.Context) (scannerInfo, error) { details.SetSizeHint(fInfo.Size()) } } else { - f.log.Errorf("Failed to read metadata from file '%v'", nextPath) + f.log.Errorf("Failed to read metadata from file '%v': %v", nextPath, err) } scanner, err := f.scannerCtor.Create(file, func(ctx context.Context, err error) error { diff --git a/internal/impl/io/input_file_size_test.go b/internal/impl/io/input_file_size_test.go index 6e4aa56d9..6d861cdbb 100644 --- a/internal/impl/io/input_file_size_test.go +++ b/internal/impl/io/input_file_size_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package io_test diff --git a/internal/impl/pure/scanner_decompress.go b/internal/impl/pure/scanner_decompress.go index d651e4a36..f8cda4181 100644 --- a/internal/impl/pure/scanner_decompress.go +++ b/internal/impl/pure/scanner_decompress.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package pure @@ -63,6 +63,14 @@ func (c *decompressScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc if !ok { cRdr = io.NopCloser(dRdr) } + // A size hint, if present, describes the compressed stream, but the child + // reads the decompressed one, whose length is unknown. Strip the hint on a + // copy rather than mislead the child (or mutate the caller's details). + if details.SizeHint() != 0 { + trimmed := *details + trimmed.SetSizeHint(0) + details = &trimmed + } return c.child.Create(cRdr, aFn, details) } diff --git a/internal/impl/pure/scanner_decompress_test.go b/internal/impl/pure/scanner_decompress_test.go index b307bb399..c2fea1ffb 100644 --- a/internal/impl/pure/scanner_decompress_test.go +++ b/internal/impl/pure/scanner_decompress_test.go @@ -1,11 +1,15 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package pure_test import ( + "bytes" + "context" "encoding/hex" + "io" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/redpanda-data/benthos/v4/internal/component/scanner/testutil" @@ -32,3 +36,95 @@ test: testutil.ScannerTestSuite(t, rdr, nil, inputBytes, "hello", "world", "this", "is", "compressed") } + +// hintCapture records the source details observed by the capture_hint_test +// scanner, so a test can assert on what a wrapping scanner propagated. +var hintCapture = struct { + name string + size int64 +}{} + +func init() { + service.MustRegisterBatchScannerCreator("capture_hint_test", + service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})), + func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) { + return &captureHintScannerCreator{}, nil + }) +} + +type captureHintScannerCreator struct{} + +func (c *captureHintScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { + hintCapture.name = details.Name() + hintCapture.size = details.SizeHint() + return service.AutoAggregateBatchScannerAcks(&captureHintScanner{r: rdr}, aFn), nil +} + +func (c *captureHintScannerCreator) Close(context.Context) error { return nil } + +type captureHintScanner struct { + r io.ReadCloser +} + +func (c *captureHintScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) { + if c.r == nil { + return nil, io.EOF + } + b, err := io.ReadAll(c.r) + if err != nil { + return nil, err + } + _ = c.r.Close() + c.r = nil + return service.MessageBatch{service.NewMessage(b)}, nil +} + +func (c *captureHintScanner) Close(ctx context.Context) error { + if c.r == nil { + return nil + } + return c.r.Close() +} + +// TestDecompressScannerStripsSizeHint asserts that a size hint, which +// describes the compressed stream, is not forwarded to the child scanner +// reading the decompressed one, while the rest of the details survive. +func TestDecompressScannerStripsSizeHint(t *testing.T) { + confSpec := service.NewConfigSpec().Field(service.NewScannerField("test")) + pConf, err := confSpec.ParseYAML(` +test: + decompress: + algorithm: gzip + into: + capture_hint_test: {} +`, nil) + require.NoError(t, err) + + rdr, err := pConf.FieldScanner("test") + require.NoError(t, err) + + inputBytes, err := hex.DecodeString("1f8b080000096e8800ff001e00e1ff68656c6c6f58776f726c64587468697358697358636f6d7072657373656403009104d92d1e000000") + require.NoError(t, err) + + details := service.NewScannerSourceDetails() + details.SetName("compressed.gz") + details.SetSizeHint(int64(len(inputBytes))) + + strm, err := rdr.Create(io.NopCloser(bytes.NewReader(inputBytes)), func(ctx context.Context, err error) error { + return nil + }, details) + require.NoError(t, err) + + m, _, err := strm.NextBatch(t.Context()) + require.NoError(t, err) + require.Len(t, m, 1) + + mBytes, err := m[0].AsBytes() + require.NoError(t, err) + assert.Equal(t, "helloXworldXthisXisXcompressed", string(mBytes)) + + assert.Equal(t, "compressed.gz", hintCapture.name, "name should survive the decompress layer") + assert.Zero(t, hintCapture.size, "compressed size hint should not reach the child scanner") + + require.NoError(t, strm.Close(t.Context())) +} diff --git a/internal/impl/pure/scanner_to_the_end.go b/internal/impl/pure/scanner_to_the_end.go index 2a92bd8d5..35d61a725 100644 --- a/internal/impl/pure/scanner_to_the_end.go +++ b/internal/impl/pure/scanner_to_the_end.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package pure @@ -37,13 +37,9 @@ func toTheEndScannerCreatorFromParsed(conf *service.ParsedConfig) (s *toTheEndSc type toTheEndScannerCreator struct{} func (l *toTheEndScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { - // The size hint is used to pre-allocate the read buffer only, and is absent + // The size hint is used to pre-allocate the read buffer only, and is zero // for sources of unknown length. - var sizeHint int64 - if details != nil { - sizeHint = details.SizeHint() - } - return service.AutoAggregateBatchScannerAcks(&toTheEndScanner{r: rdr, sizeHint: sizeHint}, aFn), nil + return service.AutoAggregateBatchScannerAcks(&toTheEndScanner{r: rdr, sizeHint: details.SizeHint()}, aFn), nil } func (l *toTheEndScannerCreator) Close(context.Context) error { @@ -55,6 +51,16 @@ type toTheEndScanner struct { sizeHint int64 } +// maxPreallocHint caps the capacity pre-allocated from a size hint. The hint +// originates from external data (typically a Stat) and can be wildly wrong: +// procfs files report multi-terabyte sizes, and a custom filesystem can return +// anything, including values for which make() panics outright (above the +// runtime's allocation limit, or overflowing hint+1). io.ReadAll bounded +// memory by growing incrementally, so pre-allocation must not turn a bogus +// hint into an enormous up-front allocation. Content beyond the clamp still +// reads correctly, paying only the incremental growth it always used to. +const maxPreallocHint = 1 << 30 + // readAllHinted is io.ReadAll with a starting capacity hint. // // Semantics are identical to io.ReadAll; the hint is purely an optimisation. @@ -62,14 +68,18 @@ type toTheEndScanner struct { // paying the copy only for the excess; if smaller, the read ends early. This // matters because a file can be appended to between Stat and read. // +// A hint of zero or below means the size is unknown, in which case io.ReadAll +// is used directly: its growth strategy starts at 512 bytes, which beats +// growing from an empty buffer. +// // The +1 on capacity is deliberate: the loop can only Read while cap > len, so // a buffer sized exactly to the content would find len == cap on the final // iteration and grow once more, reintroducing the reallocation this avoids. func readAllHinted(r io.Reader, hint int64) ([]byte, error) { - if hint < 0 { - hint = 0 + if hint <= 0 { + return io.ReadAll(r) } - buf := make([]byte, 0, hint+1) + buf := make([]byte, 0, min(hint, maxPreallocHint)+1) for { if len(buf) == cap(buf) { buf = append(buf, 0)[:len(buf)] diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go index b0e88e02f..9b92d0d4b 100644 --- a/internal/impl/pure/scanner_to_the_end_internal_test.go +++ b/internal/impl/pure/scanner_to_the_end_internal_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package pure @@ -6,6 +6,7 @@ import ( "bytes" "fmt" "io" + "math" "math/rand" "testing" @@ -110,6 +111,29 @@ func TestReadAllHintedExactHintDoesNotRealloc(t *testing.T) { } } +// TestReadAllHintedHugeHintDoesNotPanic asserts that an absurdly large hint — +// a procfs file reporting terabytes, a custom filesystem returning garbage, or +// a value for which make(0, hint+1) would panic outright — is clamped rather +// than trusted. Previously io.ReadAll simply ignored such values by growing +// incrementally, so this must never become a new failure mode. +func TestReadAllHintedHugeHintDoesNotPanic(t *testing.T) { + content := randomBytes(4096) + + for _, hint := range []int64{ + maxPreallocHint + 1, + 1 << 40, + 1 << 62, + math.MaxInt64, + } { + t.Run(fmt.Sprintf("hint=%v", hint), func(t *testing.T) { + act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, hint) + require.NoError(t, err) + assert.Equal(t, content, act) + assert.LessOrEqual(t, cap(act), maxPreallocHint+1, "hint was not clamped") + }) + } +} + // TestReadAllHintedGrowsWhenHintTooSmall asserts the buffer still grows to fit // content larger than the hint, the case where a file is appended to between // Stat and read. @@ -172,3 +196,20 @@ func BenchmarkToTheEndReadAllHinted(b *testing.B) { } } } + +// BenchmarkToTheEndReadAllUnhinted guards the no-hint path (every source that +// isn't a regular file), which must stay on par with io.ReadAll rather than +// growing a buffer from scratch. +func BenchmarkToTheEndReadAllUnhinted(b *testing.B) { + b.ReportAllocs() + b.SetBytes(benchSize) + for b.Loop() { + buf, err := readAllHinted(&repeatReader{remaining: benchSize, chunk: 32 * 1024}, 0) + if err != nil { + b.Fatal(err) + } + if len(buf) != benchSize { + b.Fatalf("unexpected length %v", len(buf)) + } + } +} diff --git a/internal/impl/pure/scanner_to_the_end_test.go b/internal/impl/pure/scanner_to_the_end_test.go index d58dfcfdd..490be8f4a 100644 --- a/internal/impl/pure/scanner_to_the_end_test.go +++ b/internal/impl/pure/scanner_to_the_end_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package pure_test diff --git a/public/service/codec/scanner.go b/public/service/codec/scanner.go index 99e3bd82f..7ad708268 100644 --- a/public/service/codec/scanner.go +++ b/public/service/codec/scanner.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package codec @@ -118,13 +118,11 @@ type codecRPublic struct { } func (r *codecRPublic) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (DeprecatedFallbackStream, error) { - sDetails := service.NewScannerSourceDetails() - if details != nil { - sDetails.SetName(details.Name()) - sDetails.SetSizeHint(details.SizeHint()) - } - - os, err := r.newCtor.Create(rdr, aFn, sDetails) + // The details are passed through wholesale rather than copied field by + // field: a copy silently drops any field this layer doesn't know about + // (which is how SizeHint was originally lost here), and the downstream + // Create is nil-safe. + os, err := r.newCtor.Create(rdr, aFn, details) if err != nil { return nil, err } diff --git a/public/service/scanner.go b/public/service/scanner.go index 955d78551..c18951b70 100644 --- a/public/service/scanner.go +++ b/public/service/scanner.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package service @@ -33,22 +33,32 @@ func (r *ScannerSourceDetails) SetName(name string) { } // Name returns a filename (or other equivalent name of the source), or an -// empty string if it has not been set. +// empty string if it has not been set. It is safe to call on a nil receiver, +// as details are optional. func (r *ScannerSourceDetails) Name() string { + if r == nil { + return "" + } return r.details.Name } // SetSizeHint sets the total size in bytes of the source to details. This is a // hint only, used by scanner implementations in order to pre-allocate buffers, // and must never be relied upon for correctness as the underlying source may -// change between it being measured and read. +// change between it being measured and read. A zero size is indistinguishable +// from the hint not being set at all, so implementations treat zero as +// unknown. func (r *ScannerSourceDetails) SetSizeHint(size int64) { r.details.SizeHint = size } -// SizeHint returns the total size in bytes of the source, or zero if it has not -// been set. +// SizeHint returns the total size in bytes of the source, or zero if it is +// unknown or has not been set. It is safe to call on a nil receiver, as +// details are optional. func (r *ScannerSourceDetails) SizeHint() int64 { + if r == nil { + return 0 + } return r.details.SizeHint } diff --git a/public/service/scanner_test.go b/public/service/scanner_test.go new file mode 100644 index 000000000..0fc1c9367 --- /dev/null +++ b/public/service/scanner_test.go @@ -0,0 +1,19 @@ +// Copyright 2026 Redpanda Data, Inc. + +package service + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestScannerSourceDetailsNilGetters asserts that the getters are safe to call +// on a nil receiver, returning zero values. Details are optional throughout +// the scanner APIs, so a nil pointer must behave like empty details. +func TestScannerSourceDetailsNilGetters(t *testing.T) { + var details *ScannerSourceDetails + + assert.Empty(t, details.Name()) + assert.Zero(t, details.SizeHint()) +} From 3746024ca55b2cd0df682d70b27c9c354e3aa229 Mon Sep 17 00:00:00 2001 From: rockdatasrl001 <167617386+rockdatasrl001@users.noreply.github.com> Date: Wed, 19 Aug 2026 18:21:29 +0200 Subject: [PATCH 3/8] Update internal/impl/io/input_file_size_test.go Co-authored-by: Joseph Woodward --- internal/impl/io/input_file_size_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/io/input_file_size_test.go b/internal/impl/io/input_file_size_test.go index 6e4aa56d9..6d861cdbb 100644 --- a/internal/impl/io/input_file_size_test.go +++ b/internal/impl/io/input_file_size_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package io_test From f501ba81705661a366694b9fa4a2a1be5bc7c4dc Mon Sep 17 00:00:00 2001 From: rockdatasrl001 <167617386+rockdatasrl001@users.noreply.github.com> Date: Wed, 19 Aug 2026 18:21:41 +0200 Subject: [PATCH 4/8] Update internal/impl/pure/scanner_to_the_end_internal_test.go Co-authored-by: Joseph Woodward --- internal/impl/pure/scanner_to_the_end_internal_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go index b0e88e02f..fff33471a 100644 --- a/internal/impl/pure/scanner_to_the_end_internal_test.go +++ b/internal/impl/pure/scanner_to_the_end_internal_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. package pure From 171939fce120ed83e4953b2cc1cfe6eaa790d9eb Mon Sep 17 00:00:00 2001 From: Ashley Jeffs Date: Wed, 19 Aug 2026 17:31:40 +0100 Subject: [PATCH 5/8] scanner: polish size hint edges from review Follow-ups from a second review pass over the branch: - readAllHinted copies the result down when a hint overestimated the content (source truncated between Stat and read, or a misreporting filesystem), so the returned message no longer pins the whole pre-allocated array for its lifetime. An accurate hint leaves cap at len+1 and never pays the copy. - The SizeHint contract docs now state the wrapping-scanner invariant that decompress previously enforced only as inline code: a wrapper that transforms stream length must clear the hint before forwarding details. Setter docs also note they panic on a nil receiver, unlike the getters. - The capture-scanner test scaffolding duplicated between the file input and decompress scanner tests is consolidated into internal/component/scanner/testutil, with the capture callbacks now consistently mutex-guarded. --- CHANGELOG.md | 2 +- internal/component/scanner/interface.go | 5 ++ .../scanner/testutil/capture_scanner.go | 57 +++++++++++++++++++ internal/impl/io/input_file_size_test.go | 44 +------------- internal/impl/pure/scanner_decompress_test.go | 54 +++++------------- internal/impl/pure/scanner_to_the_end.go | 9 +++ .../pure/scanner_to_the_end_internal_test.go | 14 +++++ public/service/scanner.go | 7 +++ 8 files changed, 109 insertions(+), 83 deletions(-) create mode 100644 internal/component/scanner/testutil/capture_scanner.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ba5d2a5a..12de6b34f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ All notable changes to this project will be documented in this file. ### Fixed -- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned: sources of unknown size retain the exact `io.ReadAll` behaviour, pre-allocation from a hint is capped at 1 GiB to guard against wildly wrong sizes, and the `decompress` scanner strips the hint before it reaches its child, as it describes the compressed stream. (@rockdatasrl001) +- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned: sources of unknown size retain the exact `io.ReadAll` behaviour, pre-allocation from a hint is capped at 1 GiB to guard against wildly wrong sizes, a hint that overestimates the content does not pin the excess capacity for the message's lifetime, and the `decompress` scanner strips the hint before it reaches its child, as it describes the compressed stream. (@rockdatasrl001) ## 4.77.0 - 2026-07-30 diff --git a/internal/component/scanner/interface.go b/internal/component/scanner/interface.go index 6ee2158cc..58ede9a4b 100644 --- a/internal/component/scanner/interface.go +++ b/internal/component/scanner/interface.go @@ -32,6 +32,11 @@ type SourceDetails struct { // implementations can pre-allocate buffers, and must never be relied upon // for correctness as the underlying source may change between it being // measured and read. + // + // A scanner that wraps a child scanner and transforms the length of the + // stream (such as decompress) must clear the hint before forwarding + // details, as it describes the stream the wrapper reads, not the one the + // child does. SizeHint int64 } diff --git a/internal/component/scanner/testutil/capture_scanner.go b/internal/component/scanner/testutil/capture_scanner.go new file mode 100644 index 000000000..3edf7a9a3 --- /dev/null +++ b/internal/component/scanner/testutil/capture_scanner.go @@ -0,0 +1,57 @@ +// Copyright 2026 Redpanda Data, Inc. + +package testutil + +import ( + "context" + "io" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// MustRegisterDetailsCaptureScanner registers a scanner plugin under name that +// invokes fn with the source details each stream is created with, then reads +// the stream to the end and delivers it as a single message. It is intended +// for tests asserting on the details an input or wrapping scanner propagates. +func MustRegisterDetailsCaptureScanner(name string, fn func(*service.ScannerSourceDetails)) { + service.MustRegisterBatchScannerCreator(name, + service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})), + func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) { + return &captureScannerCreator{fn: fn}, nil + }) +} + +type captureScannerCreator struct { + fn func(*service.ScannerSourceDetails) +} + +func (c *captureScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { + c.fn(details) + return service.AutoAggregateBatchScannerAcks(&captureScanner{r: rdr}, aFn), nil +} + +func (c *captureScannerCreator) Close(context.Context) error { return nil } + +type captureScanner struct { + r io.ReadCloser +} + +func (c *captureScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) { + if c.r == nil { + return nil, io.EOF + } + b, err := io.ReadAll(c.r) + if err != nil { + return nil, err + } + _ = c.r.Close() + c.r = nil + return service.MessageBatch{service.NewMessage(b)}, nil +} + +func (c *captureScanner) Close(ctx context.Context) error { + if c.r == nil { + return nil + } + return c.r.Close() +} diff --git a/internal/impl/io/input_file_size_test.go b/internal/impl/io/input_file_size_test.go index 6d861cdbb..9db41e276 100644 --- a/internal/impl/io/input_file_size_test.go +++ b/internal/impl/io/input_file_size_test.go @@ -3,9 +3,7 @@ package io_test import ( - "context" "fmt" - "io" "os" "path/filepath" "sync" @@ -15,6 +13,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + scannertestutil "github.com/redpanda-data/benthos/v4/internal/component/scanner/testutil" "github.com/redpanda-data/benthos/v4/internal/component/testutil" "github.com/redpanda-data/benthos/v4/internal/manager/mock" "github.com/redpanda-data/benthos/v4/internal/message" @@ -29,48 +28,11 @@ var sizeCapture = struct { }{sizes: map[string]int64{}} func init() { - service.MustRegisterBatchScannerCreator("capture_size_test", - service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})), - func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) { - return &captureSizeScannerCreator{}, nil - }) -} - -type captureSizeScannerCreator struct{} - -func (c *captureSizeScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { - if details != nil { + scannertestutil.MustRegisterDetailsCaptureScanner("capture_size_test", func(details *service.ScannerSourceDetails) { sizeCapture.Lock() sizeCapture.sizes[filepath.Base(details.Name())] = details.SizeHint() sizeCapture.Unlock() - } - return service.AutoAggregateBatchScannerAcks(&captureSizeScanner{r: rdr}, aFn), nil -} - -func (c *captureSizeScannerCreator) Close(context.Context) error { return nil } - -type captureSizeScanner struct { - r io.ReadCloser -} - -func (c *captureSizeScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) { - if c.r == nil { - return nil, io.EOF - } - b, err := io.ReadAll(c.r) - if err != nil { - return nil, err - } - _ = c.r.Close() - c.r = nil - return service.MessageBatch{service.NewMessage(b)}, nil -} - -func (c *captureSizeScanner) Close(ctx context.Context) error { - if c.r == nil { - return nil - } - return c.r.Close() + }) } // TestFileInputPropagatesSize asserts that the file input reads Size() from the diff --git a/internal/impl/pure/scanner_decompress_test.go b/internal/impl/pure/scanner_decompress_test.go index c2fea1ffb..a1405a7e0 100644 --- a/internal/impl/pure/scanner_decompress_test.go +++ b/internal/impl/pure/scanner_decompress_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/hex" "io" + "sync" "testing" "github.com/stretchr/testify/assert" @@ -40,50 +41,18 @@ test: // hintCapture records the source details observed by the capture_hint_test // scanner, so a test can assert on what a wrapping scanner propagated. var hintCapture = struct { + sync.Mutex name string size int64 }{} func init() { - service.MustRegisterBatchScannerCreator("capture_hint_test", - service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})), - func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) { - return &captureHintScannerCreator{}, nil - }) -} - -type captureHintScannerCreator struct{} - -func (c *captureHintScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { - hintCapture.name = details.Name() - hintCapture.size = details.SizeHint() - return service.AutoAggregateBatchScannerAcks(&captureHintScanner{r: rdr}, aFn), nil -} - -func (c *captureHintScannerCreator) Close(context.Context) error { return nil } - -type captureHintScanner struct { - r io.ReadCloser -} - -func (c *captureHintScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) { - if c.r == nil { - return nil, io.EOF - } - b, err := io.ReadAll(c.r) - if err != nil { - return nil, err - } - _ = c.r.Close() - c.r = nil - return service.MessageBatch{service.NewMessage(b)}, nil -} - -func (c *captureHintScanner) Close(ctx context.Context) error { - if c.r == nil { - return nil - } - return c.r.Close() + testutil.MustRegisterDetailsCaptureScanner("capture_hint_test", func(details *service.ScannerSourceDetails) { + hintCapture.Lock() + hintCapture.name = details.Name() + hintCapture.size = details.SizeHint() + hintCapture.Unlock() + }) } // TestDecompressScannerStripsSizeHint asserts that a size hint, which @@ -123,8 +92,11 @@ test: require.NoError(t, err) assert.Equal(t, "helloXworldXthisXisXcompressed", string(mBytes)) - assert.Equal(t, "compressed.gz", hintCapture.name, "name should survive the decompress layer") - assert.Zero(t, hintCapture.size, "compressed size hint should not reach the child scanner") + hintCapture.Lock() + capturedName, capturedSize := hintCapture.name, hintCapture.size + hintCapture.Unlock() + assert.Equal(t, "compressed.gz", capturedName, "name should survive the decompress layer") + assert.Zero(t, capturedSize, "compressed size hint should not reach the child scanner") require.NoError(t, strm.Close(t.Context())) } diff --git a/internal/impl/pure/scanner_to_the_end.go b/internal/impl/pure/scanner_to_the_end.go index 35d61a725..a0523fba3 100644 --- a/internal/impl/pure/scanner_to_the_end.go +++ b/internal/impl/pure/scanner_to_the_end.go @@ -90,6 +90,15 @@ func readAllHinted(r io.Reader, hint int64) ([]byte, error) { if err == io.EOF { err = nil } + // A hint that overestimated the content (the source shrank between + // being measured and read, or the measurement was wrong) would + // otherwise leave the caller pinning the whole pre-allocated array + // for as long as it holds the returned bytes; copy down when the + // waste exceeds the content. An accurate hint leaves cap at len+1 + // and never pays this copy. + if int64(cap(buf)) > int64(len(buf))*2 { + buf = append(make([]byte, 0, len(buf)), buf...) + } return buf, err } } diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go index 9b92d0d4b..56bc98b1e 100644 --- a/internal/impl/pure/scanner_to_the_end_internal_test.go +++ b/internal/impl/pure/scanner_to_the_end_internal_test.go @@ -134,6 +134,20 @@ func TestReadAllHintedHugeHintDoesNotPanic(t *testing.T) { } } +// TestReadAllHintedOverHintReleasesExcess asserts that a hint which grossly +// overestimates the content (a file truncated between Stat and read, or a +// misreporting filesystem) does not leave the returned slice pinning the +// whole pre-allocated array for the lifetime of the message. +func TestReadAllHintedOverHintReleasesExcess(t *testing.T) { + const size = 4096 + content := randomBytes(size) + + act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, 1<<20) + require.NoError(t, err) + assert.Equal(t, content, act) + assert.Less(t, cap(act), size*2+1, "over-hinted buffer was not copied down") +} + // TestReadAllHintedGrowsWhenHintTooSmall asserts the buffer still grows to fit // content larger than the hint, the case where a file is appended to between // Stat and read. diff --git a/public/service/scanner.go b/public/service/scanner.go index c18951b70..3c539545d 100644 --- a/public/service/scanner.go +++ b/public/service/scanner.go @@ -28,6 +28,7 @@ func NewScannerSourceDetails() *ScannerSourceDetails { } // SetName sets a filename (or other equivalent name of the source) to details. +// Unlike the getters, calling this on a nil receiver panics. func (r *ScannerSourceDetails) SetName(name string) { r.details.Name = name } @@ -48,6 +49,12 @@ func (r *ScannerSourceDetails) Name() string { // change between it being measured and read. A zero size is indistinguishable // from the hint not being set at all, so implementations treat zero as // unknown. +// +// A scanner that wraps a child scanner and transforms the length of the stream +// (such as decompress) must clear the hint before forwarding details, as it +// describes the stream the wrapper reads, not the one the child does. +// +// Unlike the getters, calling this on a nil receiver panics. func (r *ScannerSourceDetails) SetSizeHint(size int64) { r.details.SizeHint = size } From 32f7dafd7879a9b9312c987ceb231e3f4a500ac2 Mon Sep 17 00:00:00 2001 From: Ashley Jeffs Date: Wed, 19 Aug 2026 19:50:21 +0100 Subject: [PATCH 6/8] scanner: tune size hint allocation behaviour Third pass from review, tightening the memory behaviour of the hinted read and reconciling the contract docs with the in-tree wrappers: - The pre-allocation clamp drops from 1 GiB to 128 MiB: large enough that typical whole-stream messages keep the single-allocation benefit, small enough that a bogus Stat can't spike a constrained deployment before a byte is read. - Growth past the hint (source grew after being measured, or content beyond the clamp) now doubles explicitly instead of relying on append's ~1.25x strategy, keeping cumulative allocation linear in the content size much like io.ReadAll's chunking, with the same +1 trick so content ending exactly at a doubled capacity doesn't force one more doubling just to observe EOF. - The pre-allocation is floored at io.ReadAll's 512-byte starting size, so a tiny stale hint doesn't begin with pathologically small reads. - The copy-down of excess capacity now only runs on success (the caller discards the buffer on error) and only when the waste exceeds 64 KiB, where it's worth the copy. - The file input ignores negative Stat sizes from misbehaving custom filesystems, keeping the documented zero-means-unknown contract. - The SizeHint contract now distinguishes wrappers that materially transform stream length (decompress: must clear the hint) from those trimming a bounded few bytes (skip_bom: may forward it, and now says so at the forwarding site). - The details-capture test scanner owns its synchronised state and returns an accessor, rather than each test hand-rolling globals. - Changelog notes the 128 MiB cap and that deprecated string codec configs are unaffected. --- CHANGELOG.md | 2 +- internal/component/scanner/interface.go | 10 ++- .../scanner/testutil/capture_scanner.go | 34 ++++++-- internal/impl/io/input_file.go | 8 +- internal/impl/io/input_file_size_test.go | 29 +++---- internal/impl/pure/scanner_decompress_test.go | 28 ++----- internal/impl/pure/scanner_skip_bom.go | 3 + internal/impl/pure/scanner_to_the_end.go | 79 ++++++++++++------- .../pure/scanner_to_the_end_internal_test.go | 2 +- public/service/scanner.go | 9 ++- 10 files changed, 117 insertions(+), 87 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 12de6b34f..0c7f15182 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ All notable changes to this project will be documented in this file. ### Fixed -- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned: sources of unknown size retain the exact `io.ReadAll` behaviour, pre-allocation from a hint is capped at 1 GiB to guard against wildly wrong sizes, a hint that overestimates the content does not pin the excess capacity for the message's lifetime, and the `decompress` scanner strips the hint before it reaches its child, as it describes the compressed stream. (@rockdatasrl001) +- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned: sources of unknown size retain the exact `io.ReadAll` behaviour, pre-allocation from a hint is capped at 128 MiB to guard against wildly wrong sizes, a hint that overestimates the content does not pin the excess capacity for the message's lifetime, and the `decompress` scanner strips the hint before it reaches its child, as it describes the compressed stream. Deprecated string `codec` configs are unaffected. (@rockdatasrl001) ## 4.77.0 - 2026-07-30 diff --git a/internal/component/scanner/interface.go b/internal/component/scanner/interface.go index 58ede9a4b..8577875fa 100644 --- a/internal/component/scanner/interface.go +++ b/internal/component/scanner/interface.go @@ -33,10 +33,12 @@ type SourceDetails struct { // for correctness as the underlying source may change between it being // measured and read. // - // A scanner that wraps a child scanner and transforms the length of the - // stream (such as decompress) must clear the hint before forwarding - // details, as it describes the stream the wrapper reads, not the one the - // child does. + // A scanner that wraps a child scanner and materially transforms the + // length of the stream (such as decompress) must clear the hint before + // forwarding details, as it describes the stream the wrapper reads, not + // the one the child does. A wrapper that shortens the stream by only a + // bounded few bytes (such as skip_bom) may forward the hint unchanged, as + // it remains a tight upper bound. SizeHint int64 } diff --git a/internal/component/scanner/testutil/capture_scanner.go b/internal/component/scanner/testutil/capture_scanner.go index 3edf7a9a3..053dc7d7b 100644 --- a/internal/component/scanner/testutil/capture_scanner.go +++ b/internal/component/scanner/testutil/capture_scanner.go @@ -5,20 +5,44 @@ package testutil import ( "context" "io" + "slices" + "sync" "github.com/redpanda-data/benthos/v4/public/service" ) +// CapturedDetails is a snapshot of the source details a capture scanner +// observed at stream creation. +type CapturedDetails struct { + Name string + SizeHint int64 +} + // MustRegisterDetailsCaptureScanner registers a scanner plugin under name that -// invokes fn with the source details each stream is created with, then reads -// the stream to the end and delivers it as a single message. It is intended -// for tests asserting on the details an input or wrapping scanner propagates. -func MustRegisterDetailsCaptureScanner(name string, fn func(*service.ScannerSourceDetails)) { +// records the source details each stream is created with, then reads the +// stream to the end and delivers it as a single message. The returned function +// reports every capture so far, oldest first, and is safe for concurrent use. +// It is intended for tests asserting on the details an input or wrapping +// scanner propagates. +func MustRegisterDetailsCaptureScanner(name string) func() []CapturedDetails { + var mut sync.Mutex + var captured []CapturedDetails + service.MustRegisterBatchScannerCreator(name, service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})), func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) { - return &captureScannerCreator{fn: fn}, nil + return &captureScannerCreator{fn: func(details *service.ScannerSourceDetails) { + mut.Lock() + captured = append(captured, CapturedDetails{Name: details.Name(), SizeHint: details.SizeHint()}) + mut.Unlock() + }}, nil }) + + return func() []CapturedDetails { + mut.Lock() + defer mut.Unlock() + return slices.Clone(captured) + } } type captureScannerCreator struct { diff --git a/internal/impl/io/input_file.go b/internal/impl/io/input_file.go index 278452418..297e759ad 100644 --- a/internal/impl/io/input_file.go +++ b/internal/impl/io/input_file.go @@ -168,9 +168,11 @@ func (f *fileConsumer) getReader(ctx context.Context) (scannerInfo, error) { modTimeUTC = fInfo.ModTime().UTC() // Only regular files have a meaningful size to pre-allocate against, a - // FIFO, device or directory does not. - if fInfo.Mode().IsRegular() { - details.SetSizeHint(fInfo.Size()) + // FIFO, device or directory does not. The positivity check keeps the + // documented SizeHint contract (zero means unknown, never negative) + // even when a custom filesystem misreports a size. + if s := fInfo.Size(); s > 0 && fInfo.Mode().IsRegular() { + details.SetSizeHint(s) } } else { f.log.Errorf("Failed to read metadata from file '%v': %v", nextPath, err) diff --git a/internal/impl/io/input_file_size_test.go b/internal/impl/io/input_file_size_test.go index 9db41e276..035c4ceae 100644 --- a/internal/impl/io/input_file_size_test.go +++ b/internal/impl/io/input_file_size_test.go @@ -6,7 +6,6 @@ import ( "fmt" "os" "path/filepath" - "sync" "testing" "time" @@ -17,23 +16,11 @@ import ( "github.com/redpanda-data/benthos/v4/internal/component/testutil" "github.com/redpanda-data/benthos/v4/internal/manager/mock" "github.com/redpanda-data/benthos/v4/internal/message" - "github.com/redpanda-data/benthos/v4/public/service" ) -// sizeCapture records the source details observed by a scanner, so that a test -// can assert on what the file input actually propagated. -var sizeCapture = struct { - sync.Mutex - sizes map[string]int64 -}{sizes: map[string]int64{}} - -func init() { - scannertestutil.MustRegisterDetailsCaptureScanner("capture_size_test", func(details *service.ScannerSourceDetails) { - sizeCapture.Lock() - sizeCapture.sizes[filepath.Base(details.Name())] = details.SizeHint() - sizeCapture.Unlock() - }) -} +// capturedSizes reports the source details observed by the capture_size_test +// scanner, so that a test can assert on what the file input propagated. +var capturedSizes = scannertestutil.MustRegisterDetailsCaptureScanner("capture_size_test") // TestFileInputPropagatesSize asserts that the file input reads Size() from the // Stat it already performs and passes it to the scanner, for regular files. @@ -67,10 +54,12 @@ file: assert.Equal(t, content, string(tran.Payload.Get(0).AsBytes())) require.NoError(t, tran.Ack(t.Context(), nil)) - sizeCapture.Lock() - got := sizeCapture.sizes["sized.txt"] - sizeCapture.Unlock() - + var got int64 + for _, c := range capturedSizes() { + if filepath.Base(c.Name) == "sized.txt" { + got = c.SizeHint + } + } assert.Equal(t, int64(len(content)), got, "file input should propagate the size it already obtained from Stat") } diff --git a/internal/impl/pure/scanner_decompress_test.go b/internal/impl/pure/scanner_decompress_test.go index a1405a7e0..0320151db 100644 --- a/internal/impl/pure/scanner_decompress_test.go +++ b/internal/impl/pure/scanner_decompress_test.go @@ -7,7 +7,6 @@ import ( "context" "encoding/hex" "io" - "sync" "testing" "github.com/stretchr/testify/assert" @@ -38,22 +37,9 @@ test: testutil.ScannerTestSuite(t, rdr, nil, inputBytes, "hello", "world", "this", "is", "compressed") } -// hintCapture records the source details observed by the capture_hint_test +// capturedHints reports the source details observed by the capture_hint_test // scanner, so a test can assert on what a wrapping scanner propagated. -var hintCapture = struct { - sync.Mutex - name string - size int64 -}{} - -func init() { - testutil.MustRegisterDetailsCaptureScanner("capture_hint_test", func(details *service.ScannerSourceDetails) { - hintCapture.Lock() - hintCapture.name = details.Name() - hintCapture.size = details.SizeHint() - hintCapture.Unlock() - }) -} +var capturedHints = testutil.MustRegisterDetailsCaptureScanner("capture_hint_test") // TestDecompressScannerStripsSizeHint asserts that a size hint, which // describes the compressed stream, is not forwarded to the child scanner @@ -92,11 +78,11 @@ test: require.NoError(t, err) assert.Equal(t, "helloXworldXthisXisXcompressed", string(mBytes)) - hintCapture.Lock() - capturedName, capturedSize := hintCapture.name, hintCapture.size - hintCapture.Unlock() - assert.Equal(t, "compressed.gz", capturedName, "name should survive the decompress layer") - assert.Zero(t, capturedSize, "compressed size hint should not reach the child scanner") + captures := capturedHints() + require.NotEmpty(t, captures) + last := captures[len(captures)-1] + assert.Equal(t, "compressed.gz", last.Name, "name should survive the decompress layer") + assert.Zero(t, last.SizeHint, "compressed size hint should not reach the child scanner") require.NoError(t, strm.Close(t.Context())) } diff --git a/internal/impl/pure/scanner_skip_bom.go b/internal/impl/pure/scanner_skip_bom.go index 05c1b8a91..1b6bb4d2a 100644 --- a/internal/impl/pure/scanner_skip_bom.go +++ b/internal/impl/pure/scanner_skip_bom.go @@ -45,6 +45,9 @@ type ssbScannerCreator struct { } func (c *ssbScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) { + // The size hint is deliberately forwarded unchanged: stripping a BOM + // shortens the stream by at most 4 bytes, so the hint remains a tight + // upper bound, which the SizeHint contract permits. return c.child.Create(skipBOM(rdr), aFn, details) } diff --git a/internal/impl/pure/scanner_to_the_end.go b/internal/impl/pure/scanner_to_the_end.go index a0523fba3..fe7f8279d 100644 --- a/internal/impl/pure/scanner_to_the_end.go +++ b/internal/impl/pure/scanner_to_the_end.go @@ -51,38 +51,62 @@ type toTheEndScanner struct { sizeHint int64 } -// maxPreallocHint caps the capacity pre-allocated from a size hint. The hint -// originates from external data (typically a Stat) and can be wildly wrong: -// procfs files report multi-terabyte sizes, and a custom filesystem can return -// anything, including values for which make() panics outright (above the -// runtime's allocation limit, or overflowing hint+1). io.ReadAll bounded -// memory by growing incrementally, so pre-allocation must not turn a bogus -// hint into an enormous up-front allocation. Content beyond the clamp still -// reads correctly, paying only the incremental growth it always used to. -const maxPreallocHint = 1 << 30 +const ( + // maxPreallocHint caps the capacity pre-allocated from a size hint. The + // hint originates from external data (typically a Stat) and can be wildly + // wrong: procfs files report multi-terabyte sizes, and a custom + // filesystem can return anything, including values for which make() + // panics outright (above the runtime's allocation limit, or overflowing + // hint+1). io.ReadAll bounded memory by growing incrementally, so + // pre-allocation must not turn a bogus hint into a memory spike that can + // take out a constrained deployment before a single byte is read. 128 MiB + // covers typical whole-stream messages while bounding the blast radius of + // a wrong hint; content beyond the clamp reads correctly, growing by + // doubling. + maxPreallocHint = 128 << 20 + + // minPreallocHint floors the pre-allocated capacity at io.ReadAll's own + // starting size, so a hint that underestimates wildly (a file appended to + // after being measured) doesn't begin with pathologically small reads. + minPreallocHint = 512 + + // maxReturnWaste is the largest gap between the returned buffer's + // capacity and its content that readAllHinted will leave in place. Beyond + // it the content is copied down, as callers hold the returned bytes for + // the lifetime of the message and would otherwise pin the excess. Below + // it a copy costs more than the memory it reclaims. + maxReturnWaste = 64 << 10 +) // readAllHinted is io.ReadAll with a starting capacity hint. // -// Semantics are identical to io.ReadAll; the hint is purely an optimisation. -// If the source is larger than the hint the buffer grows exactly as before, -// paying the copy only for the excess; if smaller, the read ends early. This -// matters because a file can be appended to between Stat and read. -// -// A hint of zero or below means the size is unknown, in which case io.ReadAll -// is used directly: its growth strategy starts at 512 bytes, which beats -// growing from an empty buffer. +// The bytes and error returned are identical to io.ReadAll for every hint +// value; the hint only shapes allocation. A hint of zero or below means the +// size is unknown, in which case io.ReadAll is used directly. // -// The +1 on capacity is deliberate: the loop can only Read while cap > len, so -// a buffer sized exactly to the content would find len == cap on the final -// iteration and grow once more, reintroducing the reallocation this avoids. +// An accurate hint reads the content into a single allocation returned as-is: +// the +1 on capacity is deliberate, as the loop can only Read while +// cap > len, so a buffer sized exactly to the content would find len == cap +// on the final iteration and grow once more, reintroducing the reallocation +// this avoids. A hint that falls short (the source grew after being measured, +// or exceeds the clamp) grows by doubling, keeping cumulative allocation +// linear in the content size much like io.ReadAll's own chunking. A buffer +// returned with meaningfully more capacity than content (the source shrank, +// or the measurement was wrong) is copied down first, so a wrong hint can't +// pin excess memory for the lifetime of the returned bytes. func readAllHinted(r io.Reader, hint int64) ([]byte, error) { if hint <= 0 { return io.ReadAll(r) } - buf := make([]byte, 0, min(hint, maxPreallocHint)+1) + buf := make([]byte, 0, max(min(hint, maxPreallocHint)+1, minPreallocHint)) for { if len(buf) == cap(buf) { - buf = append(buf, 0)[:len(buf)] + // The +1 serves the same purpose as on the initial allocation: + // content ending exactly at a doubled capacity can observe EOF in + // the spare byte rather than forcing one more doubling. + grown := make([]byte, len(buf), 2*cap(buf)+1) + copy(grown, buf) + buf = grown } n, err := r.Read(buf[len(buf):cap(buf)]) buf = buf[:len(buf)+n] @@ -90,13 +114,10 @@ func readAllHinted(r io.Reader, hint int64) ([]byte, error) { if err == io.EOF { err = nil } - // A hint that overestimated the content (the source shrank between - // being measured and read, or the measurement was wrong) would - // otherwise leave the caller pinning the whole pre-allocated array - // for as long as it holds the returned bytes; copy down when the - // waste exceeds the content. An accurate hint leaves cap at len+1 - // and never pays this copy. - if int64(cap(buf)) > int64(len(buf))*2 { + // The error path skips the copy-down: the only caller discards + // the buffer when err != nil, and io.ReadAll's contract covers + // contents and error, not capacity. + if err == nil && cap(buf)-len(buf) > maxReturnWaste { buf = append(make([]byte, 0, len(buf)), buf...) } return buf, err diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go index 56bc98b1e..4215df715 100644 --- a/internal/impl/pure/scanner_to_the_end_internal_test.go +++ b/internal/impl/pure/scanner_to_the_end_internal_test.go @@ -106,7 +106,7 @@ func TestReadAllHintedExactHintDoesNotRealloc(t *testing.T) { assert.Equal(t, content, act) assert.Len(t, act, size) - assert.Equal(t, size+1, cap(act), "buffer was reallocated despite an exact hint") + assert.Equal(t, max(size+1, minPreallocHint), cap(act), "buffer was reallocated despite an exact hint") }) } } diff --git a/public/service/scanner.go b/public/service/scanner.go index 3c539545d..27c8d6d47 100644 --- a/public/service/scanner.go +++ b/public/service/scanner.go @@ -50,9 +50,12 @@ func (r *ScannerSourceDetails) Name() string { // from the hint not being set at all, so implementations treat zero as // unknown. // -// A scanner that wraps a child scanner and transforms the length of the stream -// (such as decompress) must clear the hint before forwarding details, as it -// describes the stream the wrapper reads, not the one the child does. +// A scanner that wraps a child scanner and materially transforms the length +// of the stream (such as decompress) must clear the hint before forwarding +// details, as it describes the stream the wrapper reads, not the one the +// child does. A wrapper that shortens the stream by only a bounded few bytes +// (such as skip_bom) may forward the hint unchanged, as it remains a tight +// upper bound. // // Unlike the getters, calling this on a nil receiver panics. func (r *ScannerSourceDetails) SetSizeHint(size int64) { From 4d7c4ed0115f62ed7e1a7666322ad7f06c2309f9 Mon Sep 17 00:00:00 2001 From: rockdatasrl001 Date: Wed, 19 Aug 2026 21:05:07 +0200 Subject: [PATCH 7/8] fix(scanner): floor the size hint at io.ReadAll's default and guard overflow Addresses review feedback from @josephwoodward. The to_the_end scanner is also driven by inputs that pass no size hint (socket, stdin, http_client). io.ReadAll starts from a 512-byte buffer, so a small no-hint read previously grew up from a single byte and made more allocations than the io.ReadAll it replaced. Floor the hint at 512 (io.ReadAll's initial capacity) so those paths keep io.ReadAll's profile while large hints still pre-size exactly. SetSizeHint is exported, so the hint is untrusted: also clamp it below math.MaxInt64 so the +1 capacity can never overflow to a negative value. The floor/clamp is factored into a small hintedCap function so both behaviours (including the overflow branch, which can't be reached through an actual read) are unit tested. Benchmark is unchanged; make fmt, make lint and make test still pass. --- internal/impl/pure/scanner_to_the_end.go | 37 +++++++++--- .../pure/scanner_to_the_end_internal_test.go | 60 +++++++++++++++++-- 2 files changed, 83 insertions(+), 14 deletions(-) diff --git a/internal/impl/pure/scanner_to_the_end.go b/internal/impl/pure/scanner_to_the_end.go index 2a92bd8d5..00c967081 100644 --- a/internal/impl/pure/scanner_to_the_end.go +++ b/internal/impl/pure/scanner_to_the_end.go @@ -5,6 +5,7 @@ package pure import ( "context" "io" + "math" "github.com/redpanda-data/benthos/v4/public/service" ) @@ -55,21 +56,41 @@ type toTheEndScanner struct { sizeHint int64 } +// defaultReadAllCap matches the initial buffer capacity io.ReadAll uses. Sources +// without a useful size hint (sockets, stdin, and other unbounded streams that +// route through this scanner) therefore keep io.ReadAll's allocation profile +// rather than growing up from a single byte. +const defaultReadAllCap = 512 + +// hintedCap converts a caller-supplied size hint into the capacity to +// pre-allocate. +// +// The hint is floored at defaultReadAllCap so that a small, absent, or negative +// hint never allocates worse than io.ReadAll would, and capped just below +// math.MaxInt64 so the +1 below cannot overflow to a negative capacity — +// SetSizeHint is exported, so the hint is untrusted input. +// +// The +1 is deliberate: the read loop can only Read while cap > len, so a buffer +// sized to exactly the content would find len == cap on the final iteration and +// grow once more, reintroducing the reallocation this avoids. +func hintedCap(hint int64) int64 { + if hint < defaultReadAllCap { + hint = defaultReadAllCap + } + if hint > math.MaxInt64-1 { + hint = math.MaxInt64 - 1 + } + return hint + 1 +} + // readAllHinted is io.ReadAll with a starting capacity hint. // // Semantics are identical to io.ReadAll; the hint is purely an optimisation. // If the source is larger than the hint the buffer grows exactly as before, // paying the copy only for the excess; if smaller, the read ends early. This // matters because a file can be appended to between Stat and read. -// -// The +1 on capacity is deliberate: the loop can only Read while cap > len, so -// a buffer sized exactly to the content would find len == cap on the final -// iteration and grow once more, reintroducing the reallocation this avoids. func readAllHinted(r io.Reader, hint int64) ([]byte, error) { - if hint < 0 { - hint = 0 - } - buf := make([]byte, 0, hint+1) + buf := make([]byte, 0, hintedCap(hint)) for { if len(buf) == cap(buf) { buf = append(buf, 0)[:len(buf)] diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go index fff33471a..6190cdecf 100644 --- a/internal/impl/pure/scanner_to_the_end_internal_test.go +++ b/internal/impl/pure/scanner_to_the_end_internal_test.go @@ -6,6 +6,7 @@ import ( "bytes" "fmt" "io" + "math" "math/rand" "testing" @@ -90,13 +91,15 @@ type errReader struct{ err error } func (e *errReader) Read([]byte) (int, error) { return 0, e.err } -// TestReadAllHintedExactHintDoesNotRealloc asserts that an accurate hint -// results in exactly one allocation, by checking the returned buffer still has -// the capacity it was created with. This is the test that catches a regression -// of the +1 subtlety: sizing the buffer to exactly the content length would -// find len == cap on the final iteration and grow once more. +// TestReadAllHintedExactHintDoesNotRealloc asserts that an accurate hint at or +// above the default floor results in exactly one allocation, by checking the +// returned buffer still has the capacity it was created with. This is the test +// that catches a regression of the +1 subtlety: sizing the buffer to exactly the +// content length would find len == cap on the final iteration and grow once more. func TestReadAllHintedExactHintDoesNotRealloc(t *testing.T) { - for _, size := range []int{1, 512, 4096, 1 << 20} { + // Sizes are >= defaultReadAllCap so the floor does not apply and the exact + // hint is honoured; the sub-floor case is covered by TestReadAllHintedFloor. + for _, size := range []int{defaultReadAllCap, 4096, 1 << 20} { t.Run(fmt.Sprintf("size=%v", size), func(t *testing.T) { content := randomBytes(size) @@ -110,6 +113,51 @@ func TestReadAllHintedExactHintDoesNotRealloc(t *testing.T) { } } +// TestHintedCap asserts the hint is floored at the io.ReadAll default and capped +// so the +1 never overflows to a negative capacity. The overflow branch cannot +// be exercised through readAllHinted (allocating a MaxInt64 buffer would fail), +// so the clamping is unit tested in isolation here. +func TestHintedCap(t *testing.T) { + for _, test := range []struct { + name string + hint int64 + exp int64 + }{ + {"negative is floored", -5, defaultReadAllCap + 1}, + {"zero is floored", 0, defaultReadAllCap + 1}, + {"below default is floored", 100, defaultReadAllCap + 1}, + {"exactly default", defaultReadAllCap, defaultReadAllCap + 1}, + {"above default is honoured", 4096, 4097}, + {"max int64 does not overflow", math.MaxInt64, math.MaxInt64}, + {"max int64 minus one does not overflow", math.MaxInt64 - 1, math.MaxInt64}, + } { + t.Run(test.name, func(t *testing.T) { + got := hintedCap(test.hint) + assert.Equal(t, test.exp, got) + assert.Positive(t, got, "capacity must stay positive (no overflow)") + }) + } +} + +// TestReadAllHintedFloor asserts that a small, absent, or negative hint still +// allocates the io.ReadAll-sized buffer rather than growing up from a single +// byte, so no-hint callers (sockets, stdin, etc.) keep io.ReadAll's profile. +func TestReadAllHintedFloor(t *testing.T) { + const contentLen = 10 // < defaultReadAllCap, so it fits without any growth + for _, hint := range []int64{-5, 0, 1, 100, defaultReadAllCap} { + t.Run(fmt.Sprintf("hint=%v", hint), func(t *testing.T) { + content := randomBytes(contentLen) + + act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 4}, hint) + require.NoError(t, err) + + assert.Equal(t, content, act) + assert.Equal(t, defaultReadAllCap+1, cap(act), + "sub-floor hint should allocate the default-sized buffer") + }) + } +} + // TestReadAllHintedGrowsWhenHintTooSmall asserts the buffer still grows to fit // content larger than the hint, the case where a file is appended to between // Stat and read. From 4798ead190645f550d45ac9aabac43b10bb0f7c0 Mon Sep 17 00:00:00 2001 From: Ashley Jeffs Date: Thu, 20 Aug 2026 13:22:12 +0100 Subject: [PATCH 8/8] Use go.mod for ci tooling version --- .github/workflows/test.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 27d0712ec..09e3e38d2 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -22,7 +22,7 @@ jobs: - name: Install Go uses: actions/setup-go@v6 with: - go-version: stable + go-version-file: 'go.mod' - name: Install Task uses: ./.github/actions/setup-task @@ -49,7 +49,7 @@ jobs: - name: Install Go uses: actions/setup-go@v6 with: - go-version: stable + go-version-file: 'go.mod' - name: Set version env variables run: |