wire: bound complete streaming payload growth

### wire/common.go
@@ -43,21 +43,58 @@ func readBytes(r io.Reader, count uint64) ([]byte, error) {
 		return []byte{}, nil
 	}
 
-	remaining, known := readerRemaining(r)
-	if known && count <= remaining {
+	available, known := readerRemaining(r)
+	if known && count <= available {
 		result := make([]byte, count)
 		n, err := io.ReadFull(r, result)
 		return result[:n], err
 	}
 
-	var result bytes.Buffer
-	result.Grow(int(min(count, defaultReadBufferSize)))
-	n, err := io.CopyN(&result, r, int64(count))
-	if err == io.EOF && n > 0 {
-		err = io.ErrUnexpectedEOF
+	// Read into bounded chunks rather than using io.CopyN with a
+	// bytes.Buffer. Buffer.ReadFrom reserves more space before its final EOF
+	// check, which can grow the buffer beyond count after a complete read.
+	// Each chunk below is capped by the remaining payload and is allocated
+	// only after the preceding bytes arrive.
+	remaining := count
+	chunkSize := uint64(defaultReadBufferSize)
+
+	// The current payload maximum needs fewer than 16 geometrically sized
+	// chunks, so the chunk metadata does not need to grow while reading.
+	chunks := make([][]byte, 0, 16)
+	totalRead := 0
+	for remaining > 0 {
+		chunkSize = min(chunkSize, remaining)
+		chunk := make([]byte, int(chunkSize))
+		n := 0
+		var err error
+		for n < len(chunk) && err == nil {
+			var bytesRead int
+			bytesRead, err = r.Read(chunk[n:])
+			n += bytesRead
+		}
+
+		chunks = append(chunks, chunk[:n])
+		totalRead += n
+		remaining -= uint64(n)
+		if err != nil && remaining > 0 {
+			if err == io.EOF && totalRead > 0 {
+				err = io.ErrUnexpectedEOF
+			}
+			if len(chunks) == 1 {
+				return chunks[0], err
+			}
+
+			return bytes.Join(chunks, nil), err
+		}
+
+		chunkSize *= 2
+	}
+
+	if len(chunks) == 1 {
+		return chunks[0], nil
 	}
 
-	return result.Bytes(), err
+	return bytes.Join(chunks, nil), nil
 }
 
 // readerRemaining returns the number of buffered bytes exposed by the concrete

### wire/decode_budget_bench_test.go
@@ -6,6 +6,7 @@ package wire
 
 import (
 	"bytes"
+	"os"
 	"testing"
 )
 
@@ -23,6 +24,56 @@ func benchmarkCountPayload(b *testing.B, prefix []byte, count uint64) []byte {
 	return payload.Bytes()
 }
 
+func benchmarkCompleteStreamingMessage(b *testing.B, msg Message) {
+	b.Helper()
+
+	var encoded bytes.Buffer
+	_, err := WriteMessageWithEncodingN(
+		&encoded, msg, ProtocolVersion, MainNet, WitnessEncoding,
+	)
+	if err != nil {
+		b.Fatal(err)
+	}
+	payload := encoded.Bytes()
+
+	b.SetBytes(int64(len(payload)))
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		reader := &readSizeRecorder{
+			reader: bytes.NewReader(payload),
+		}
+		_, _, _, err := ReadMessageWithEncodingN(
+			reader, ProtocolVersion, MainNet, WitnessEncoding,
+		)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+// BenchmarkCompleteStreamingMessage measures complete framed messages read
+// from a reader that does not expose its remaining length.
+func BenchmarkCompleteStreamingMessage(b *testing.B) {
+	b.Run("ping", func(b *testing.B) {
+		benchmarkCompleteStreamingMessage(b, NewMsgPing(1))
+	})
+
+	blockBytes, err := os.ReadFile(
+		"testdata/block-00000000000000000021868c2cefc52a480d173c849412fe81c4e5ab806f94ab.blk",
+	)
+	if err != nil {
+		b.Fatal(err)
+	}
+	var block MsgBlock
+	if err := block.Deserialize(bytes.NewReader(blockBytes)); err != nil {
+		b.Fatal(err)
+	}
+	b.Run("block", func(b *testing.B) {
+		benchmarkCompleteStreamingMessage(b, &block)
+	})
+}
+
 // BenchmarkTruncatedDecodeBudget measures allocation when a length or element
 // count is present without the bytes needed to satisfy it.
 func BenchmarkTruncatedDecodeBudget(b *testing.B) {

### wire/decode_budget_test.go
@@ -6,6 +6,7 @@ package wire
 
 import (
 	"bytes"
+	"errors"
 	"io"
 	"testing"
 
@@ -304,11 +305,66 @@ func TestTruncatedVariableBytesBudget(t *testing.T) {
 	}
 }
 
+// TestCompleteStreamingBytesCapacity ensures a complete opaque read returns an
+// exactly sized payload instead of retaining capacity beyond its length.
+func TestCompleteStreamingBytesCapacity(t *testing.T) {
+	payload := bytes.Repeat(
+		[]byte{0x01}, defaultReadBufferSize+1,
+	)
+	reader := &readSizeRecorder{reader: bytes.NewReader(payload)}
+	result, err := readBytes(reader, uint64(len(payload)))
+	if err != nil {
+		t.Fatalf("unable to read complete payload: %v", err)
+	}
+	if !bytes.Equal(result, payload) {
+		t.Fatal("decoded payload differs from input")
+	}
+	if cap(result) != len(result) {
+		t.Fatalf("payload capacity is %d, want %d", cap(result), len(result))
+	}
+}
+
+// TestStreamingBytesPreservesBoundaryError ensures an error returned with a
+// full internal chunk is not discarded while the payload remains incomplete.
+func TestStreamingBytesPreservesBoundaryError(t *testing.T) {
+	errBoundary := errors.New("chunk boundary")
+	reader := &boundaryErrorReader{
+		payload: bytes.Repeat([]byte{0x01}, defaultReadBufferSize),
+		err:     errBoundary,
+	}
+	result, err := readBytes(reader, defaultReadBufferSize+1)
+	if !errors.Is(err, errBoundary) {
+		t.Fatalf("unexpected read error: got %v, want %v", err, errBoundary)
+	}
+	if len(result) != defaultReadBufferSize {
+		t.Fatalf("payload length is %d, want %d", len(result),
+			defaultReadBufferSize)
+	}
+	if reader.reads != 1 {
+		t.Fatalf("reader called %d times, want 1", reader.reads)
+	}
+}
+
 type readSizeRecorder struct {
 	reader io.Reader
 	max    int
 }
 
+type boundaryErrorReader struct {
+	payload []byte
+	err     error
+	reads   int
+}
+
+func (r *boundaryErrorReader) Read(p []byte) (int, error) {
+	r.reads++
+	if r.reads > 1 {
+		return 0, io.EOF
+	}
+
+	return copy(p, r.payload), r.err
+}
+
 type inflatedLenReader struct {
 	reader io.Reader
 	length int