perf(storer): reuse buffers and eliminate per-chunk allocations in reserve sample

pkg/file/joiner/joiner_test.go

@@ -1396,6 +1396,18 @@ func (c *chunkStore) Get(_ context.Context, addr swarm.Address) (swarm.Chunk, er
 	return chunk, nil
 }
 
+func (c *chunkStore) GetInto(ctx context.Context, addr swarm.Address, buf []byte) (int, error) {
+	ch, err := c.Get(ctx, addr)
+	if err != nil {
+		return 0, err
+	}
+	data := ch.Data()
+	if len(buf) < len(data) {
+		return 0, fmt.Errorf("chunk store: buffer too small: %d < %d", len(buf), len(data))
+	}
+	return copy(buf, data), nil
+}
+
 func (c *chunkStore) Put(_ context.Context, ch swarm.Chunk) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()

pkg/storage/chunkstore.go

@@ -21,6 +21,14 @@ type Getter interface {
 	Get(context.Context, swarm.Address) (swarm.Chunk, error)
 }
 
+// GetterInto is like Getter but reads chunk data into a caller-provided buffer,
+// avoiding per-call allocations. len(buf) must be at least the chunk size;
+// GetInto never writes past len(buf). Returns the number of bytes read into
+// buf; callers use buf[:n].
+type GetterInto interface {
+	GetInto(ctx context.Context, addr swarm.Address, buf []byte) (int, error)
+}
+
 // Putter is the interface that wraps the basic Put method.
 type Putter interface {
 	// Put a chunk into the store alongside with its postage stamp.
@@ -73,6 +81,7 @@ type ChunkGetterDeleter interface {
 
 type ChunkStore interface {
 	Getter
+	GetterInto
 	Putter
 	Deleter
 	Hasser
@@ -84,5 +93,6 @@ type ChunkStore interface {
 
 type ReadOnlyChunkStore interface {
 	Getter
+	GetterInto
 	Hasser
 }

pkg/storage/inmemchunkstore/inmemchunkstore.go

@@ -6,6 +6,7 @@ package inmemchunkstore
 
 import (
 	"context"
+	"fmt"
 	"sync"
 
 	"github.com/ethersphere/bee/v2/pkg/storage"
@@ -39,6 +40,21 @@ func (c *ChunkStore) Get(_ context.Context, addr swarm.Address) (swarm.Chunk, er
 	return chunk.chunk, nil
 }
 
+func (c *ChunkStore) GetInto(_ context.Context, addr swarm.Address, buf []byte) (int, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	chunk, ok := c.chunks[c.key(addr)]
+	if !ok {
+		return 0, storage.ErrNotFound
+	}
+	data := chunk.chunk.Data()
+	if len(buf) < len(data) {
+		return 0, fmt.Errorf("chunk store: buffer too small: %d < %d", len(buf), len(data))
+	}
+	return copy(buf, data), nil
+}
+
 func (c *ChunkStore) Put(_ context.Context, ch swarm.Chunk) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()

pkg/storer/export_test.go

@@ -5,8 +5,10 @@
 package storer
 
 import (
+	"github.com/ethersphere/bee/v2/pkg/bmt"
 	"github.com/ethersphere/bee/v2/pkg/storer/internal/events"
 	"github.com/ethersphere/bee/v2/pkg/storer/internal/reserve"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
 )
 
 func (db *DB) Reserve() *reserve.Reserve {
@@ -35,3 +37,14 @@ func (db *DB) WaitForBgCacheWorkers() (unblock func()) {
 func DefaultOptions() *Options {
 	return defaultOptions()
 }
+
+// TransformedAddress exposes the sampler's per-chunk hashing so it can be
+// benchmarked on its own.
+//
+// The exported signature takes a swarm.Chunk and is deliberately held stable
+// even where transformedAddress itself does not, so that the same benchmark
+// source can be run against branches that shape the internal function
+// differently. Only this shim changes between them.
+func TransformedAddress(hasher bmt.Hasher, ch swarm.Chunk, chType swarm.ChunkType) (swarm.Address, error) {
+	return transformedAddress(hasher, ch.Address(), ch.Data(), chType)
+}

pkg/storer/internal/chunkstore/chunkstore.go

@@ -51,6 +51,28 @@ func Get(ctx context.Context, r storage.Reader, s storage.Sharky, addr swarm.Add
 	return readChunk(ctx, s, rIdx)
 }
 
+// GetInto reads chunk data into the provided buffer, avoiding per-call allocation.
+// Returns the number of bytes read.
+func GetInto(ctx context.Context, r storage.Reader, s storage.Sharky, addr swarm.Address, buf []byte) (int, error) {
+	rIdx := &RetrievalIndexItem{Address: addr}
+	err := r.Get(rIdx)
+	if err != nil {
+		return 0, fmt.Errorf("chunk store: failed reading retrievalIndex for address %s: %w", addr, err)
+	}
+	n := int(rIdx.Location.Length)
+	if len(buf) < n {
+		return 0, fmt.Errorf("chunk store: buffer too small: %d < %d", len(buf), n)
+	}
+	err = s.Read(ctx, rIdx.Location, buf[:n])
+	if err != nil {
+		return 0, fmt.Errorf(
+			"chunk store: failed reading location: %v for chunk %s from sharky: %w",
+			rIdx.Location, rIdx.Address, err,
+		)
+	}
+	return n, nil
+}
+
 // helper to read chunk from retrievalIndex.
 func readChunk(ctx context.Context, s storage.Sharky, rIdx *RetrievalIndexItem) (swarm.Chunk, error) {
 	buf := make([]byte, rIdx.Location.Length)

pkg/storer/internal/chunkstore/chunkstore_test.go

@@ -17,13 +17,12 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/crypto"
 	"github.com/ethersphere/bee/v2/pkg/sharky"
 	soctesting "github.com/ethersphere/bee/v2/pkg/soc/testing"
-	"github.com/ethersphere/bee/v2/pkg/storer/internal/transaction"
-
 	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/storage/inmemstore"
 	"github.com/ethersphere/bee/v2/pkg/storage/storagetest"
 	chunktest "github.com/ethersphere/bee/v2/pkg/storage/testing"
 	"github.com/ethersphere/bee/v2/pkg/storer/internal/chunkstore"
+	"github.com/ethersphere/bee/v2/pkg/storer/internal/transaction"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
@@ -497,17 +496,160 @@ type chunkStore struct {
 	sharky *sharky.Store
 }
 
-func makeStorage(t *testing.T) *chunkStore {
-	t.Helper()
+func makeStorage(tb testing.TB) *chunkStore {
+	tb.Helper()
 
 	store := inmemstore.New()
 	sharky, err := sharky.New(&memFS{Fs: afero.NewMemMapFs()}, 1, swarm.SocMaxChunkSize)
-	assert.NoError(t, err)
+	assert.NoError(tb, err)
 
-	t.Cleanup(func() {
-		assert.NoError(t, store.Close())
-		assert.NoError(t, sharky.Close())
+	tb.Cleanup(func() {
+		assert.NoError(tb, store.Close())
+		assert.NoError(tb, sharky.Close())
 	})
 
 	return &chunkStore{transaction.NewStorage(sharky, store), sharky}
 }
+
+// BenchmarkChunkStoreGet measures a single chunk read: one retrieval-index
+// lookup followed by one sharky read. It is the micro-benchmark for the read
+// path that the reserve sampler drives once per chunk.
+//
+// The two variants differ only in where the ChunkStore handle comes from.
+// "per_call" mirrors what the sampler does today, building a fresh handle for
+// every chunk; "hoisted" builds it once. The gap between them is the cost of
+// that handle alone.
+func BenchmarkChunkStoreGet(b *testing.B) {
+	ctx := context.Background()
+
+	setup := func(b *testing.B) (*chunkStore, swarm.Address) {
+		b.Helper()
+
+		st := makeStorage(b)
+		ch := chunktest.GenerateTestRandomChunk()
+
+		err := st.Run(ctx, func(s transaction.Store) error {
+			return s.ChunkStore().Put(ctx, ch)
+		})
+		if err != nil {
+			b.Fatal(err)
+		}
+
+		return st, ch.Address()
+	}
+
+	b.Run("per_call", func(b *testing.B) {
+		st, addr := setup(b)
+
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		for b.Loop() {
+			if _, err := st.ChunkStore().Get(ctx, addr); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+
+	b.Run("hoisted", func(b *testing.B) {
+		st, addr := setup(b)
+		cs := st.ChunkStore()
+
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		for b.Loop() {
+			if _, err := cs.Get(ctx, addr); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+
+	b.Run("get_into", func(b *testing.B) {
+		st, addr := setup(b)
+		cs := st.ChunkStore()
+		buf := make([]byte, swarm.SocMaxChunkSize)
+
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		for b.Loop() {
+			if _, err := cs.GetInto(ctx, addr, buf); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+}
+
+// TestGetInto checks the buffer contract of GetInto: the size check is on
+// len(buf), one full-length buffer serves both a CAC and a larger SOC via the
+// returned count, nothing is written past len(buf), and a buffer whose length
+// is too short is rejected even when its capacity would fit the chunk.
+func TestGetInto(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := makeStorage(t)
+
+	cac := chunktest.GenerateTestRandomChunk()
+	soc := soctesting.GenerateMockSOC(t, cac.Data()[swarm.SpanSize:]).Chunk()
+	if len(soc.Data()) <= len(cac.Data()) {
+		t.Fatalf("soc data (%d) should be larger than cac data (%d)", len(soc.Data()), len(cac.Data()))
+	}
+
+	for _, ch := range []swarm.Chunk{cac, soc} {
+		err := st.Run(ctx, func(s transaction.Store) error {
+			return s.ChunkStore().Put(ctx, ch)
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	cs := st.ChunkStore()
+
+	t.Run("reused buffer serves CAC and SOC", func(t *testing.T) {
+		buf := make([]byte, swarm.SocMaxChunkSize)
+		for _, ch := range []swarm.Chunk{cac, soc, cac} {
+			n, err := cs.GetInto(ctx, ch.Address(), buf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if n != len(ch.Data()) || !bytes.Equal(buf[:n], ch.Data()) {
+				t.Fatalf("chunk %s: got %d bytes, want %d with matching data", ch.Address(), n, len(ch.Data()))
+			}
+		}
+	})
+
+	t.Run("does not write past len", func(t *testing.T) {
+		backing := make([]byte, swarm.SocMaxChunkSize)
+		for i := range backing {
+			backing[i] = 0xff
+		}
+		buf := backing[:len(cac.Data())]
+		n, err := cs.GetInto(ctx, cac.Address(), buf)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if n != len(cac.Data()) || !bytes.Equal(buf, cac.Data()) {
+			t.Fatalf("got %d bytes, want %d with matching data", n, len(cac.Data()))
+		}
+		for i := n; i < len(backing); i++ {
+			if backing[i] != 0xff {
+				t.Fatalf("GetInto wrote past len(buf) at index %d", i)
+			}
+		}
+	})
+
+	t.Run("length too small", func(t *testing.T) {
+		// Capacity would fit the chunk; only the length is short.
+		buf := make([]byte, len(soc.Data())-1, swarm.SocMaxChunkSize)
+		n, err := cs.GetInto(ctx, soc.Address(), buf)
+		if err == nil {
+			t.Fatal("expected error for buffer shorter than the chunk")
+		}
+		if n != 0 {
+			t.Fatalf("got %d bytes, want 0", n)
+		}
+	})
+}

pkg/storer/internal/transaction/transaction.go

@@ -225,6 +225,13 @@ func (c *chunkStoreTrx) Get(ctx context.Context, addr swarm.Address) (ch swarm.C
 	return ch, err
 }
 
+func (c *chunkStoreTrx) GetInto(ctx context.Context, addr swarm.Address, buf []byte) (n int, err error) {
+	defer handleMetric("chunkstore_get", c.metrics)(&err)
+	unlock := c.lock(addr)
+	defer unlock()
+	return chunkstore.GetInto(ctx, c.indexStore, c.sharkyTrx, addr, buf)
+}
+
 func (c *chunkStoreTrx) Has(ctx context.Context, addr swarm.Address) (_ bool, err error) {
 	defer handleMetric("chunkstore_has", c.metrics)(&err)
 	unlock := c.lock(addr)