From c1d1f5fa529f3f9bc0e96053ec6e610fcd918752 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 14 Sep 2026 11:57:55 +0200 Subject: [PATCH 01/10] test(storer): guard sample chunk data and add read-path benchmarks Sampling is about to start reading chunks into a buffer that each worker reuses. Nothing today checks that the bytes handed back in a SampleItem are still the bytes of that chunk, so a reused buffer would silently hand the redistribution proof the contents of some later chunk. assertValidSample now checks two things for every item: that ChunkData still reproduces ChunkAddress, and that no two items share a backing array. Every existing sample test picks both up. Also add the rulers for the work that follows. BenchmarkReserveSample1k keeps its name and behaviour so the recorded baseline stays comparable; its body moves to a helper that BenchmarkReserveSample10k reuses over a ten times larger reserve. BenchmarkChunkStoreGet measures a single chunk read, split into a variant that builds the ChunkStore handle per call as the sampler does today and one that hoists it, so the cost of the handle alone is visible. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KFUseFQ8rhp6N9X6YS7pbq --- .../internal/chunkstore/chunkstore_test.go | 67 +++++++++++++++++-- pkg/storer/sample_test.go | 64 +++++++++++++++++- 2 files changed, 123 insertions(+), 8 deletions(-) diff --git a/pkg/storer/internal/chunkstore/chunkstore_test.go b/pkg/storer/internal/chunkstore/chunkstore_test.go index 970e92df9b8..00047a16881 100644 --- a/pkg/storer/internal/chunkstore/chunkstore_test.go +++ b/pkg/storer/internal/chunkstore/chunkstore_test.go @@ -497,17 +497,72 @@ 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) + } + } + }) +} diff --git a/pkg/storer/sample_test.go b/pkg/storer/sample_test.go index 16b65d81a09..f9976061d23 100644 --- a/pkg/storer/sample_test.go +++ b/pkg/storer/sample_test.go @@ -14,6 +14,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/bmt" "github.com/ethersphere/bee/v2/pkg/cac" "github.com/ethersphere/bee/v2/pkg/postage" + "github.com/ethersphere/bee/v2/pkg/soc" postagetesting "github.com/ethersphere/bee/v2/pkg/postage/testing" chunk "github.com/ethersphere/bee/v2/pkg/storage/testing" @@ -302,6 +303,9 @@ func assertValidSample(t *testing.T, sample storer.Sample, minRadius uint8, anch assertSampleItem(item, i) } + assertSampleDataIntact(t, sample) + assertSampleDataNotShared(t, sample) + // Assert that transformed addresses are in ascending order for i := 0; i < len(sample.Items)-1; i++ { if sample.Items[i].TransformedAddress.Compare(sample.Items[i+1].TransformedAddress) != -1 { @@ -310,6 +314,44 @@ func assertValidSample(t *testing.T, sample storer.Sample, minRadius uint8, anch } } +// assertSampleDataIntact checks that every item's ChunkData still reproduces its +// ChunkAddress. +// +// The sampler currently hands out the buffer that the chunk store allocated for +// each chunk, so the data is trivially intact. That stops being true the moment +// chunks are read into a buffer that the worker reuses: unless the bytes handed +// to a SampleItem are copied out, every item except the last one a worker +// touched carries the contents of some later chunk instead. +func assertSampleDataIntact(t *testing.T, sample storer.Sample) { + t.Helper() + + for i, item := range sample.Items { + ch := swarm.NewChunk(item.ChunkAddress, item.ChunkData) + if !cac.Valid(ch) && !soc.Valid(ch) { + t.Fatalf("sample item [%d]: chunk data does not reproduce address %s", i, item.ChunkAddress) + } + } +} + +// assertSampleDataNotShared checks that no two items are backed by the same +// array. It catches a reused read buffer even in the case where the surviving +// contents happen to stay valid for one of the aliased items. +func assertSampleDataNotShared(t *testing.T, sample storer.Sample) { + t.Helper() + + seen := make(map[*byte]int, len(sample.Items)) + for i, item := range sample.Items { + if len(item.ChunkData) == 0 { + continue + } + first := &item.ChunkData[0] + if j, ok := seen[first]; ok { + t.Fatalf("sample items [%d] and [%d] share one backing array", j, i) + } + seen[first] = i + } +} + // TestSampleVectorCAC is a deterministic test vector that verifies the chunk // address and transformed address produced by MakeSampleUsingChunks for a // single hardcoded CAC chunk and anchor. It guards against regressions in the @@ -420,11 +462,27 @@ func assertSampleNoErrors(t *testing.T, sample storer.Sample) { // method, including DB iteration, chunk loading, stamp validation, and sample // assembly. func BenchmarkReserveSample1k(b *testing.B) { - const chunkCountPerPO = 100 + benchmarkReserveSample(b, 100) +} + +// BenchmarkReserveSample10k is BenchmarkReserveSample1k over a reserve ten +// times the size. The per-chunk costs of the sampler are linear in the number +// of chunks iterated, so a change that only removes a fixed overhead reads the +// same at both sizes while a change to a per-chunk allocation does not. The +// larger reserve is also where garbage collection starts to show. +func BenchmarkReserveSample10k(b *testing.B) { + benchmarkReserveSample(b, 1000) +} + +// benchmarkReserveSample fills a reserve with chunkCountPerPO chunks in each of +// the first maxPO proximity orders and then samples it repeatedly. +func benchmarkReserveSample(b *testing.B, chunkCountPerPO int) { + b.Helper() + const maxPO = 10 baseAddr := swarm.RandAddress(b) - opts := dbTestOps(baseAddr, 5000, nil, nil, time.Second) + opts := dbTestOps(baseAddr, 5*chunkCountPerPO*maxPO, nil, nil, time.Second) opts.ValidStamp = func(ch swarm.Chunk) (swarm.Chunk, error) { return ch, nil } st, err := diskStorer(b, opts)() @@ -450,6 +508,8 @@ func BenchmarkReserveSample1k(b *testing.B) { anchor = swarm.RandAddressAt(b, baseAddr, int(radius)).Bytes() ) + b.ResetTimer() + for b.Loop() { _, err := st.ReserveSample(context.TODO(), anchor, radius, timeVar, nil) if err != nil { From abc65aca5674b296fb2d0c93b5cea066fe2ed4d4 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 14 Sep 2026 11:58:59 +0200 Subject: [PATCH 02/10] perf(storer): build the sampler's chunk store handle once per worker db.ChunkStore() builds three objects every time it is called, and the sampler called it once per chunk. On a testnet node that is 2.3 million handles per round, thrown away immediately. Hoist it to one per worker. The handle is deliberately not shared across workers: the read-only chunk store makes no thread-safety promise, and three allocations per worker is already nothing next to three per chunk. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KFUseFQ8rhp6N9X6YS7pbq --- pkg/storer/sample.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/pkg/storer/sample.go b/pkg/storer/sample.go index f4b6f8139e3..de5bb86c8b4 100644 --- a/pkg/storer/sample.go +++ b/pkg/storer/sample.go @@ -127,6 +127,11 @@ func (db *DB) ReserveSample( g.Go(safe.RunFunc(db.logger, "storer-sample-worker", func() error { wstat := SampleStats{} hasher := bmt.NewPrefixHasher(anchor) + // One handle per worker rather than one per chunk: building it + // allocates, and the sampler asks for a chunk millions of times per + // round. It is not shared between workers because the read-only + // chunk store makes no thread-safety promise. + chunkStore := db.ChunkStore() defer func() { addStats(wstat) }() @@ -147,7 +152,7 @@ func (db *DB) ReserveSample( chunkLoadStart := time.Now() - chunk, err := db.ChunkStore().Get(ctx, chItem.Address) + chunk, err := chunkStore.Get(ctx, chItem.Address) chunkLoadDuration := time.Since(chunkLoadStart) if err != nil { From 16824419f59d9479884307d18ff8413984cab286 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 14 Sep 2026 13:19:17 +0200 Subject: [PATCH 03/10] perf(storer): read into reusable worker buffer and load sample items lazily --- pkg/file/joiner/joiner_test.go | 13 ++++ pkg/storage/chunkstore.go | 9 +++ .../inmemchunkstore/inmemchunkstore.go | 17 ++++++ pkg/storer/internal/chunkstore/chunkstore.go | 22 +++++++ .../internal/chunkstore/chunkstore_test.go | 18 +++++- .../internal/transaction/transaction.go | 7 +++ pkg/storer/sample.go | 61 +++++++++++-------- pkg/storer/sample_test.go | 3 +- 8 files changed, 122 insertions(+), 28 deletions(-) diff --git a/pkg/file/joiner/joiner_test.go b/pkg/file/joiner/joiner_test.go index dd1a7e4228e..ad7c40b8819 100644 --- a/pkg/file/joiner/joiner_test.go +++ b/pkg/file/joiner/joiner_test.go @@ -1396,6 +1396,19 @@ 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)) + } + copy(buf, data) + return len(data), nil +} + func (c *chunkStore) Put(_ context.Context, ch swarm.Chunk) error { c.mu.Lock() defer c.mu.Unlock() diff --git a/pkg/storage/chunkstore.go b/pkg/storage/chunkstore.go index 68a9d10652a..2809493e129 100644 --- a/pkg/storage/chunkstore.go +++ b/pkg/storage/chunkstore.go @@ -21,6 +21,13 @@ 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. The buffer must be large enough to hold the chunk. +// Returns the number of bytes read into buf. +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 +80,7 @@ type ChunkGetterDeleter interface { type ChunkStore interface { Getter + GetterInto Putter Deleter Hasser @@ -84,5 +92,6 @@ type ChunkStore interface { type ReadOnlyChunkStore interface { Getter + GetterInto Hasser } diff --git a/pkg/storage/inmemchunkstore/inmemchunkstore.go b/pkg/storage/inmemchunkstore/inmemchunkstore.go index 4f0465016af..e775a6bc884 100644 --- a/pkg/storage/inmemchunkstore/inmemchunkstore.go +++ b/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,22 @@ 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)) + } + copy(buf, data) + return len(data), nil +} + func (c *ChunkStore) Put(_ context.Context, ch swarm.Chunk) error { c.mu.Lock() defer c.mu.Unlock() diff --git a/pkg/storer/internal/chunkstore/chunkstore.go b/pkg/storer/internal/chunkstore/chunkstore.go index 6d2745cc5fd..0b70134f38b 100644 --- a/pkg/storer/internal/chunkstore/chunkstore.go +++ b/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) diff --git a/pkg/storer/internal/chunkstore/chunkstore_test.go b/pkg/storer/internal/chunkstore/chunkstore_test.go index 00047a16881..d07728d321d 100644 --- a/pkg/storer/internal/chunkstore/chunkstore_test.go +++ b/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" @@ -565,4 +564,19 @@ func BenchmarkChunkStoreGet(b *testing.B) { } } }) + + 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) + } + } + }) } diff --git a/pkg/storer/internal/transaction/transaction.go b/pkg/storer/internal/transaction/transaction.go index 0f403580152..2aa1072e624 100644 --- a/pkg/storer/internal/transaction/transaction.go +++ b/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) diff --git a/pkg/storer/sample.go b/pkg/storer/sample.go index de5bb86c8b4..19e680b3d6d 100644 --- a/pkg/storer/sample.go +++ b/pkg/storer/sample.go @@ -8,6 +8,7 @@ import ( "bytes" "context" "encoding/binary" + "errors" "fmt" "math/big" "runtime" @@ -65,7 +66,7 @@ func (db *DB) ReserveSample( consensusTime uint64, minBatchBalance *big.Int, ) (Sample, error) { - g, ctx := errgroup.WithContext(ctx) + g, gCtx := errgroup.WithContext(ctx) allStats := &SampleStats{} statsLock := sync.Mutex{} @@ -111,8 +112,8 @@ func (db *DB) ReserveSample( case chunkC <- ch: stats.TotalIterated++ return false, nil - case <-ctx.Done(): - return false, ctx.Err() + case <-gCtx.Done(): + return false, gCtx.Err() } }) return err @@ -132,6 +133,7 @@ func (db *DB) ReserveSample( // round. It is not shared between workers because the read-only // chunk store makes no thread-safety promise. chunkStore := db.ChunkStore() + buf := make([]byte, swarm.SocMaxChunkSize) defer func() { addStats(wstat) }() @@ -152,7 +154,7 @@ func (db *DB) ReserveSample( chunkLoadStart := time.Now() - chunk, err := chunkStore.Get(ctx, chItem.Address) + n, err := chunkStore.GetInto(gCtx, chItem.Address, buf) chunkLoadDuration := time.Since(chunkLoadStart) if err != nil { @@ -164,7 +166,7 @@ func (db *DB) ReserveSample( wstat.ChunkLoadDuration += chunkLoadDuration taddrStart := time.Now() - taddr, err := transformedAddress(hasher, chunk, chItem.ChunkType) + taddr, err := transformedAddress(hasher, chItem.Address, buf[:n], chItem.ChunkType) if err != nil { return err } @@ -173,12 +175,11 @@ func (db *DB) ReserveSample( select { case sampleItemChan <- SampleItem{ TransformedAddress: taddr, - ChunkAddress: chunk.Address(), - ChunkData: chunk.Data(), + ChunkAddress: chItem.Address, Stamp: postage.NewStamp(chItem.BatchID, nil, nil, nil), }: - case <-ctx.Done(): - return ctx.Err() + case <-gCtx.Done(): + return gCtx.Err() } } @@ -224,6 +225,7 @@ func (db *DB) ReserveSample( // Phase 3: Assemble the sample. Here we need to assemble only the first SampleSize // no of items from the results of the 2nd phase. // In this step stamps are loaded and validated only if chunk will be added to sample. + phase3ChunkStore := db.ChunkStore() stats := SampleStats{} for item := range sampleItemChan { currentMaxAddr := swarm.EmptyAddress @@ -239,7 +241,14 @@ func (db *DB) ReserveSample( continue } - ch := swarm.NewChunk(item.ChunkAddress, item.ChunkData).WithStamp(stamp) + ch, err := phase3ChunkStore.Get(ctx, item.ChunkAddress) + if err != nil { + stats.ChunkLoadFailed++ + db.logger.Debug("failed loading chunk", "chunk_address", item.ChunkAddress, "error", err) + continue + } + + ch = ch.WithStamp(stamp) // check if the timestamp on the postage stamp is not later than the consensus time. if binary.BigEndian.Uint64(ch.Stamp().Timestamp()) > consensusTime { @@ -258,6 +267,7 @@ func (db *DB) ReserveSample( stats.ValidStampDuration += stampValidDuration item.Stamp = postage.NewStamp(stamp.BatchID(), stamp.Index(), stamp.Timestamp(), stamp.Sig()) + item.ChunkData = ch.Data() insert(item) stats.SampleInserts++ @@ -300,22 +310,25 @@ func (db *DB) batchesBelowValue(until *big.Int) (map[string]struct{}, error) { return res, err } -func transformedAddress(hasher bmt.Hasher, chunk swarm.Chunk, chType swarm.ChunkType) (swarm.Address, error) { +func transformedAddress(hasher bmt.Hasher, addr swarm.Address, data []byte, chType swarm.ChunkType) (swarm.Address, error) { switch chType { case swarm.ChunkTypeContentAddressed: - return transformedAddressCAC(hasher, chunk) + return transformedAddressCAC(hasher, data) case swarm.ChunkTypeSingleOwner: - return transformedAddressSOC(hasher, chunk) + return transformedAddressSOC(hasher, addr, data) default: return swarm.ZeroAddress, fmt.Errorf("chunk type [%v] is not valid", chType) } } -func transformedAddressCAC(hasher bmt.Hasher, chunk swarm.Chunk) (swarm.Address, error) { +func transformedAddressCAC(hasher bmt.Hasher, data []byte) (swarm.Address, error) { + if len(data) < bmt.SpanSize { + return swarm.ZeroAddress, errors.New("chunk data too short for span") + } hasher.Reset() - hasher.SetHeader(chunk.Data()[:bmt.SpanSize]) + hasher.SetHeader(data[:bmt.SpanSize]) - _, err := hasher.Write(chunk.Data()[bmt.SpanSize:]) + _, err := hasher.Write(data[bmt.SpanSize:]) if err != nil { return swarm.ZeroAddress, err } @@ -323,20 +336,20 @@ func transformedAddressCAC(hasher bmt.Hasher, chunk swarm.Chunk) (swarm.Address, return swarm.NewAddress(hasher.Sum(nil)), nil } -func transformedAddressSOC(hasher bmt.Hasher, socChunk swarm.Chunk) (swarm.Address, error) { - // Calculate transformed address from wrapped chunk - cacChunk, err := soc.UnwrapCAC(socChunk) - if err != nil { - return swarm.ZeroAddress, err +func transformedAddressSOC(hasher bmt.Hasher, socAddr swarm.Address, data []byte) (swarm.Address, error) { + if len(data) < swarm.SocMinChunkSize { + return swarm.ZeroAddress, errors.New("chunk data too short for soc") } - taddrCac, err := transformedAddressCAC(hasher, cacChunk) + cursor := swarm.HashSize + swarm.SocSignatureSize + cacData := data[cursor:] + taddrCac, err := transformedAddressCAC(hasher, cacData) if err != nil { return swarm.ZeroAddress, err } // Hash address and transformed address to make transformed address for this SOC sHasher := swarm.NewHasher() - if _, err := sHasher.Write(socChunk.Address().Bytes()); err != nil { + if _, err := sHasher.Write(socAddr.Bytes()); err != nil { return swarm.ZeroAddress, err } if _, err := sHasher.Write(taddrCac.Bytes()); err != nil { @@ -405,7 +418,7 @@ func RandSample(t *testing.T, anchor []byte) Sample { func MakeSampleUsingChunks(chunks []swarm.Chunk, anchor []byte) (Sample, error) { items := make([]SampleItem, len(chunks)) for i, ch := range chunks { - tr, err := transformedAddress(bmt.NewPrefixHasher(anchor), ch, getChunkType(ch)) + tr, err := transformedAddress(bmt.NewPrefixHasher(anchor), ch.Address(), ch.Data(), getChunkType(ch)) if err != nil { return Sample{}, err } diff --git a/pkg/storer/sample_test.go b/pkg/storer/sample_test.go index f9976061d23..bd0036004be 100644 --- a/pkg/storer/sample_test.go +++ b/pkg/storer/sample_test.go @@ -14,9 +14,8 @@ import ( "github.com/ethersphere/bee/v2/pkg/bmt" "github.com/ethersphere/bee/v2/pkg/cac" "github.com/ethersphere/bee/v2/pkg/postage" - "github.com/ethersphere/bee/v2/pkg/soc" - postagetesting "github.com/ethersphere/bee/v2/pkg/postage/testing" + "github.com/ethersphere/bee/v2/pkg/soc" chunk "github.com/ethersphere/bee/v2/pkg/storage/testing" "github.com/ethersphere/bee/v2/pkg/storer" "github.com/ethersphere/bee/v2/pkg/swarm" From 841bbafef00b15bb2418eba3f4e5c4b4f51d5cb5 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 14 Sep 2026 13:52:38 +0200 Subject: [PATCH 04/10] perf(storer): pass batch id through sampler channel to avoid stamp allocations --- pkg/storer/sample.go | 36 ++++++++++++++++++++++-------------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/pkg/storer/sample.go b/pkg/storer/sample.go index 19e680b3d6d..66a21032f3c 100644 --- a/pkg/storer/sample.go +++ b/pkg/storer/sample.go @@ -38,6 +38,12 @@ type SampleItem struct { Stamp *postage.Stamp } +type sampleItemInternal struct { + transformedAddress swarm.Address + chunkAddress swarm.Address + batchID []byte +} + type Sample struct { Stats SampleStats Items []SampleItem @@ -120,7 +126,7 @@ func (db *DB) ReserveSample( })) // Phase 2: Get the chunk data and calculate transformed hash - sampleItemChan := make(chan SampleItem, 3*workers) + sampleItemChan := make(chan sampleItemInternal, 3*workers) db.logger.Debug("reserve sampler workers", "count", workers) @@ -173,10 +179,10 @@ func (db *DB) ReserveSample( wstat.TaddrDuration += time.Since(taddrStart) select { - case sampleItemChan <- SampleItem{ - TransformedAddress: taddr, - ChunkAddress: chItem.Address, - Stamp: postage.NewStamp(chItem.BatchID, nil, nil, nil), + case sampleItemChan <- sampleItemInternal{ + transformedAddress: taddr, + chunkAddress: chItem.Address, + batchID: chItem.BatchID, }: case <-gCtx.Done(): return gCtx.Err() @@ -233,18 +239,18 @@ func (db *DB) ReserveSample( currentMaxAddr = sampleItems[len(sampleItems)-1].TransformedAddress } - if le(item.TransformedAddress, currentMaxAddr) || len(sampleItems) < SampleSize { - stamp, err := chunkstamp.LoadWithBatchID(db.storage.IndexStore(), "reserve", item.ChunkAddress, item.Stamp.BatchID()) + if le(item.transformedAddress, currentMaxAddr) || len(sampleItems) < SampleSize { + stamp, err := chunkstamp.LoadWithBatchID(db.storage.IndexStore(), "reserve", item.chunkAddress, item.batchID) if err != nil { stats.StampLoadFailed++ - db.logger.Debug("failed loading stamp", "chunk_address", item.ChunkAddress, "error", err) + db.logger.Debug("failed loading stamp", "chunk_address", item.chunkAddress, "error", err) continue } - ch, err := phase3ChunkStore.Get(ctx, item.ChunkAddress) + ch, err := phase3ChunkStore.Get(ctx, item.chunkAddress) if err != nil { stats.ChunkLoadFailed++ - db.logger.Debug("failed loading chunk", "chunk_address", item.ChunkAddress, "error", err) + db.logger.Debug("failed loading chunk", "chunk_address", item.chunkAddress, "error", err) continue } @@ -266,10 +272,12 @@ func (db *DB) ReserveSample( stampValidDuration := time.Since(stampValidStart) stats.ValidStampDuration += stampValidDuration - item.Stamp = postage.NewStamp(stamp.BatchID(), stamp.Index(), stamp.Timestamp(), stamp.Sig()) - item.ChunkData = ch.Data() - - insert(item) + insert(SampleItem{ + TransformedAddress: item.transformedAddress, + ChunkAddress: item.chunkAddress, + ChunkData: ch.Data(), + Stamp: postage.NewStamp(stamp.BatchID(), stamp.Index(), stamp.Timestamp(), stamp.Sig()), + }) stats.SampleInserts++ } } From 40f51d4617e713c908fedcf758177ef5f1619d2a Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 14 Sep 2026 18:27:41 +0200 Subject: [PATCH 05/10] test(storer): benchmark transformedAddress for CAC and SOC separately Both ReserveSample benchmarks fill their reserve with chunk.GenerateValidRandomChunkAt, which produces content-addressed chunks only. The SOC branch of transformedAddress is therefore never measured by them, and any work that branch does beyond hashing the wrapped CAC is invisible. Benchmark the function directly, one case per chunk type, through a shim in export_test.go. The shim takes a swarm.Chunk and is held to that signature on purpose so the same benchmark source can be run against branches whose internal transformedAddress is shaped differently; only the shim changes between them. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KFUseFQ8rhp6N9X6YS7pbq --- pkg/storer/export_test.go | 13 +++++++++++ pkg/storer/sample_test.go | 47 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/pkg/storer/export_test.go b/pkg/storer/export_test.go index b30b5996edd..d9b779ba97f 100644 --- a/pkg/storer/export_test.go +++ b/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) +} diff --git a/pkg/storer/sample_test.go b/pkg/storer/sample_test.go index bd0036004be..31f44364ed6 100644 --- a/pkg/storer/sample_test.go +++ b/pkg/storer/sample_test.go @@ -517,6 +517,53 @@ func benchmarkReserveSample(b *testing.B, chunkCountPerPO int) { } } +// BenchmarkTransformedAddress measures the sampler's per-chunk hashing on its +// own, separately for a content-addressed and a single owner chunk. +// +// The SOC case is the reason this exists. Both ReserveSample benchmarks build +// their reserve with chunk.GenerateValidRandomChunkAt, which produces CAC +// chunks only, so the SOC branch of transformedAddress is never measured by +// them. Any work the SOC path does over and above hashing the wrapped CAC shows +// up here and nowhere else. +func BenchmarkTransformedAddress(b *testing.B) { + anchor := []byte("swarm-test-anchor-deterministic!") + + content := make([]byte, swarm.ChunkSize) + for i := range content { + content[i] = byte(i) + } + + cacChunk, err := cac.New(content) + if err != nil { + b.Fatal(err) + } + socChunk := chunk.GenerateTestRandomSoChunk(b, cacChunk) + + for _, tc := range []struct { + name string + ch swarm.Chunk + typ swarm.ChunkType + }{ + {"cac", cacChunk, swarm.ChunkTypeContentAddressed}, + {"soc", socChunk, swarm.ChunkTypeSingleOwner}, + } { + b.Run(tc.name, func(b *testing.B) { + // One hasher reused across iterations, as a sampler worker does. + hasher := bmt.NewPrefixHasher(anchor) + + b.ReportAllocs() + b.SetBytes(int64(len(tc.ch.Data()))) + b.ResetTimer() + + for b.Loop() { + if _, err := storer.TransformedAddress(hasher, tc.ch, tc.typ); err != nil { + b.Fatal(err) + } + } + }) + } +} + // BenchmarkSampleHashing measures the time taken by MakeSampleUsingChunks to // hash a fixed set of CAC chunks. func BenchmarkSampleHashing(b *testing.B) { From 52186741b8b9ddc3c470e5a87c2dbea637a01139 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Tue, 15 Sep 2026 07:11:11 +0200 Subject: [PATCH 06/10] perf(storer): read chunk data via opaque location hints during reserve sampling Accelerate ReserveSample by bypassing LevelDB index lookups through opaque storage.ChunkLocation hints without breaking clean storage abstractions: - Introduce storage.ChunkLocation and capability interfaces (LocatingPutter, LocatingReplacer, LocatingGetterInto) in pkg/storage - Support dual-format backward-compatible ChunkBinItem deserialization (106 bytes legacy, 114 bytes with location) avoiding database migrations - Encapsulate thread-safe LocationGuard inside internal/chunkstore to prevent use-after-free race conditions when Sharky reuses freed slots - Use GetIntoLoc in reserve sampler workers with automatic fallback to standard GetInto for legacy or guarded locations --- .../2026-09-14-locating-chunkstore-design.md | 239 ++++++++++++++++++ pkg/storage/chunkstore.go | 24 ++ pkg/storage/chunkstore_test.go | 25 ++ pkg/storer/internal/cache/cache_test.go | 3 +- pkg/storer/internal/chunkstore/chunkstore.go | 88 ++++++- .../internal/chunkstore/chunkstore_test.go | 77 ++++++ pkg/storer/internal/chunkstore/guard.go | 79 ++++++ pkg/storer/internal/chunkstore/guard_test.go | 69 +++++ pkg/storer/internal/internal.go | 3 +- pkg/storer/internal/reserve/items.go | 22 +- pkg/storer/internal/reserve/items_test.go | 53 ++++ pkg/storer/internal/reserve/reserve.go | 74 ++++-- pkg/storer/internal/reserve/reserve_test.go | 79 ++++++ .../internal/transaction/transaction.go | 45 +++- pkg/storer/sample.go | 13 +- pkg/storer/storer.go | 12 +- 16 files changed, 854 insertions(+), 51 deletions(-) create mode 100644 docs/superpowers/specs/2026-09-14-locating-chunkstore-design.md create mode 100644 pkg/storage/chunkstore_test.go create mode 100644 pkg/storer/internal/chunkstore/guard.go create mode 100644 pkg/storer/internal/chunkstore/guard_test.go diff --git a/docs/superpowers/specs/2026-09-14-locating-chunkstore-design.md b/docs/superpowers/specs/2026-09-14-locating-chunkstore-design.md new file mode 100644 index 00000000000..847c8816ec1 --- /dev/null +++ b/docs/superpowers/specs/2026-09-14-locating-chunkstore-design.md @@ -0,0 +1,239 @@ +# Design Spec: Locating ChunkStore & Direct Reserve Sampling without Breaking Abstractions + +**Date:** 2026-09-14 +**Author:** Antigravity & Ljubisa Gacevic +**Status:** Approved for Implementation Planning +**Target Branch:** `perf/sample-01-hoist-chunkstore` (or feature branch derived from it) + +--- + +## 1. Problem Statement & Motivation + +During the Swarm storage incentives game (`ReserveSample`), a node computes a sample over chunks in its reserve (typically 50,000–100,000 chunks). +Currently: +1. Phase 1 iterates `ChunkBinItem`s from the reserve index. +2. Phase 2 loads chunk data via `chunkStore.GetInto(ctx, addr, buf)`. + +Under the hood, every single `chunkStore.GetInto` call performs a LevelDB lookup for `RetrievalIndexItem{Address: addr}` to discover the chunk's `sharky.Location`, and then reads the blob from Sharky. Across 100k chunks, this results in 100k random LevelDB lookups per sampling round. + +A previous experimental branch (`perf/reserve-sample-direct-sharky-read` & `...-guard`) attempted to denormalize `sharky.Location` into `ChunkBinItem` to read directly from Sharky. However, it was met with resistance because: +1. **Broken Abstractions & Leaky Interfaces:** `reserve` and `sample.go` directly imported and depended on `sharky.Location`. Runtime type assertions (`db.storage.(transaction.SharkyReader)`) bypassed the standard `storage.ChunkStore` contracts. +2. **Use-After-Free Concurrency Bug:** When chunks are evicted or replaced during sampling, Sharky reuses freed slots. Direct Sharky reads can read newly written chunks in those reused slots, corrupting sample proofs. This forced an ad-hoc `samplingGuard` with callbacks across subsystems (`onEvict`). +3. **Heavy Database Migration (`step_08`):** Adding `Location` directly broke binary deserialization of existing `ChunkBinItem`s on disk, requiring an expensive, blocking database migration across all production nodes. + +--- + +## 2. Goals & Non-Goals + +### Goals +- **Maintain Clean Abstractions:** `reserve` must NOT import or know about `sharky`. The storage abstraction must remain clean and follow standard Go interface patterns (Interface Segregation / Capability Interfaces). +- **Zero Race Conditions (Safe Sharky Slot Reuse):** Eliminate use-after-free and silent data corruption when slots are released during sampling rounds. +- **Encapsulation:** The guard mechanism tracking released slots must be completely encapsulated within `internal/chunkstore`, with zero callbacks or lifecycle tracking leaked into `reserve` or `storer`. +- **Zero-Downtime / No Mandatory Migration:** Backward-compatible binary serialization in `ChunkBinItem.Unmarshal` that seamlessly supports both legacy items (106 bytes) and new items with location (114 bytes). +- **Graceful Fallback:** If a location is zero, legacy, or belonged to a released slot, the system automatically and transparently falls back to standard LevelDB `GetInto(ctx, addr, buf)`. + +### Non-Goals +- Changing Sharky's internal storage layout or freelist behavior. +- Altering the cryptographic BMT sample calculation or consensus logic. +- Rewriting the entire storage engine. + +--- + +## 3. Architecture & Interface Design + +### 3.1 Opaque `storage.ChunkLocation` + +In `pkg/storage/chunkstore.go`, define an opaque 8-byte value: + +```go +// ChunkLocation is an opaque locator hint for accelerated chunk retrieval. +// Implementations can encode internal storage coordinates (e.g. Sharky shard/slot/length) +// into this structure. A zero ChunkLocation indicates an unset or legacy location. +type ChunkLocation [8]byte + +func (c ChunkLocation) IsZero() bool { + return c == ChunkLocation{} +} +``` + +*Note on size:* `sharky.Location` consists of `Shard uint8 (1) + Slot uint32 (4) + Length uint16 (2) = 7 bytes`. An 8-byte array accommodates this cleanly without exposing Sharky internals. + +### 3.2 Capability Interfaces in `pkg/storage` + +Following idiomatic Go practices (similar to `io.WriterTo` / `io.ReaderFrom`), define optional capability interfaces in `pkg/storage/chunkstore.go`: + +```go +// LocatingPutter is an optional capability of a Putter that returns a ChunkLocation hint. +type LocatingPutter interface { + PutLoc(ctx context.Context, ch swarm.Chunk) (ChunkLocation, error) +} + +// LocatingReplacer is an optional capability of a Replacer that returns a ChunkLocation hint. +type LocatingReplacer interface { + ReplaceLoc(ctx context.Context, ch swarm.Chunk, emplace bool) (ChunkLocation, error) +} + +// LocatingGetterInto is an optional capability of a GetterInto that uses a ChunkLocation hint. +// Implementations MUST verify the validity of the location or guard state, and fallback to +// address-based lookup if the location is unsafe, unset, or invalid. +type LocatingGetterInto interface { + GetterInto + GetIntoLoc(ctx context.Context, addr swarm.Address, loc ChunkLocation, buf []byte) (int, error) +} +``` + +### 3.3 Internal Location Serialization (`internal/chunkstore`) + +Within `pkg/storer/internal/chunkstore`, provide internal helpers to convert between `sharky.Location` and `storage.ChunkLocation`: + +```go +func locationToChunkLocation(l sharky.Location) storage.ChunkLocation { + var cl storage.ChunkLocation + cl[0] = l.Shard + binary.BigEndian.PutUint32(cl[1:5], l.Slot) + binary.BigEndian.PutUint16(cl[5:7], l.Length) + return cl +} + +func chunkLocationToLocation(cl storage.ChunkLocation) sharky.Location { + return sharky.Location{ + Shard: cl[0], + Slot: binary.BigEndian.Uint32(cl[1:5]), + Length: binary.BigEndian.Uint16(cl[5:7]), + } +} +``` + +--- + +## 4. Concurrency Guard & Slot Reuse Protection + +### 4.1 Encapsulated `LocationGuard` + +Sharky releases slots in only two places across the entire codebase: +1. `chunkstore.Delete` +2. `chunkstore.Replace` + +We introduce a thread-safe `LocationGuard` encapsulated within the chunkstore component: + +```go +type LocationGuard struct { + mu sync.RWMutex + active int32 // active sampling session counter + freed map[storage.ChunkLocation]struct{} +} + +func (g *LocationGuard) StartSession() func() { + g.mu.Lock() + defer g.mu.Unlock() + if g.active == 0 { + g.freed = make(map[storage.ChunkLocation]struct{}) + } + g.active++ + + return func() { + g.mu.Lock() + defer g.mu.Unlock() + g.active-- + if g.active == 0 { + g.freed = nil + } + } +} + +func (g *LocationGuard) MarkFreed(loc storage.ChunkLocation) { + g.mu.Lock() + defer g.mu.Unlock() + if g.active > 0 { + g.freed[loc] = struct{}{} + } +} + +func (g *LocationGuard) IsFreed(loc storage.ChunkLocation) bool { + g.mu.RLock() + defer g.mu.RUnlock() + if g.active == 0 { + return false + } + _, found := g.freed[loc] + return found +} +``` + +### 4.2 Safe Execution in `GetIntoLoc` + +When `GetIntoLoc(ctx, addr, loc, buf)` is called: +1. If `loc.IsZero()`: call standard `GetInto(ctx, addr, buf)` (LevelDB lookup). +2. If `guard.IsFreed(loc)`: the slot was released during the active sampling session! Immediately call standard `GetInto(ctx, addr, buf)` (which will either read the new location from LevelDB or return `storage.ErrNotFound`). +3. Otherwise: read directly from `sharky.Read(ctx, chunkLocationToLocation(loc), buf)`. If Sharky returns an error, fallback to `GetInto(ctx, addr, buf)`. + +Result: **100% immune to use-after-free and slot reuse race conditions.** + +--- + +## 5. Storage & Backward-Compatible Serialization + +### 5.1 `ChunkBinItem` Definition + +In `pkg/storer/internal/reserve/items.go`: + +```go +type ChunkBinItem struct { + Bin uint8 + BinID uint64 + Address swarm.Address + BatchID []byte + StampHash []byte + ChunkType swarm.ChunkType + Location storage.ChunkLocation // Opaque 8 bytes +} +``` + +### 5.2 Two-Format Unmarshal + +- Legacy item size: `1 + 8 + 32 + 32 + 1 + 32 = 106 bytes`. +- New item size: `106 + 8 = 114 bytes`. + +```go +const ( + legacyChunkBinItemSize = 106 + chunkBinItemSizeWithLoc = 114 +) + +func (c *ChunkBinItem) Marshal() ([]byte, error) { + buf := make([]byte, chunkBinItemSizeWithLoc) + // marshal standard fields (0..106) + ... + copy(buf[106:114], c.Location[:]) + return buf, nil +} + +func (c *ChunkBinItem) Unmarshal(buf []byte) error { + switch len(buf) { + case legacyChunkBinItemSize: + // Decode standard fields; c.Location remains zeroed. + return c.unmarshalLegacy(buf) + case chunkBinItemSizeWithLoc: + if err := c.unmarshalLegacy(buf[:legacyChunkBinItemSize]); err != nil { + return err + } + copy(c.Location[:], buf[legacyChunkBinItemSize:]) + return nil + default: + return errUnmarshalInvalidSize + } +} +``` + +--- + +## 6. Verification Plan + +1. **Unit Tests:** + - Test `ChunkBinItem` marshal/unmarshal with both 106-byte (legacy) and 114-byte buffers. + - Test `LocationGuard` concurrency (concurrent `MarkFreed` and `IsFreed`). + - Test `GetIntoLoc` fallback behavior when location is zero, when slot is freed, and normal path. +2. **Race Detector:** + - Run `go test -race ./pkg/storer -run TestReserveSample` under heavy parallel eviction and insertion. +3. **Benchmarks:** + - Compare `BenchmarkReserveSample1k` with populated `ChunkLocation` vs baseline. Expect significant drop in LevelDB read operations and latency. diff --git a/pkg/storage/chunkstore.go b/pkg/storage/chunkstore.go index 2809493e129..97d20e649ed 100644 --- a/pkg/storage/chunkstore.go +++ b/pkg/storage/chunkstore.go @@ -52,6 +52,30 @@ type Replacer interface { Replace(context.Context, swarm.Chunk, bool) error } +// ChunkLocation is an opaque locator hint for accelerated chunk retrieval. +type ChunkLocation [8]byte + +// IsZero reports whether the ChunkLocation is unset. +func (c ChunkLocation) IsZero() bool { + return c == ChunkLocation{} +} + +// LocatingPutter is an optional capability of a Putter that returns a ChunkLocation hint. +type LocatingPutter interface { + PutLoc(ctx context.Context, ch swarm.Chunk) (ChunkLocation, error) +} + +// LocatingReplacer is an optional capability of a Replacer that returns a ChunkLocation hint. +type LocatingReplacer interface { + ReplaceLoc(ctx context.Context, ch swarm.Chunk, emplace bool) (ChunkLocation, error) +} + +// LocatingGetterInto is an optional capability of a GetterInto that uses a ChunkLocation hint. +type LocatingGetterInto interface { + GetterInto + GetIntoLoc(ctx context.Context, addr swarm.Address, loc ChunkLocation, buf []byte) (int, error) +} + // PutterFunc type is an adapter to allow the use of // ChunkStore as Putter interface. If f is a function // with the appropriate signature, PutterFunc(f) is a diff --git a/pkg/storage/chunkstore_test.go b/pkg/storage/chunkstore_test.go new file mode 100644 index 00000000000..f3f7562bec9 --- /dev/null +++ b/pkg/storage/chunkstore_test.go @@ -0,0 +1,25 @@ +// Copyright 2024 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package storage_test + +import ( + "testing" + + "github.com/ethersphere/bee/v2/pkg/storage" +) + +func TestChunkLocation(t *testing.T) { + t.Parallel() + + var zero storage.ChunkLocation + if !zero.IsZero() { + t.Fatal("expected zero ChunkLocation to return true for IsZero") + } + + nonZero := storage.ChunkLocation{1, 2, 3} + if nonZero.IsZero() { + t.Fatal("expected non-zero ChunkLocation to return false for IsZero") + } +} diff --git a/pkg/storer/internal/cache/cache_test.go b/pkg/storer/internal/cache/cache_test.go index 580263cb6ac..e8a35b6aee5 100644 --- a/pkg/storer/internal/cache/cache_test.go +++ b/pkg/storer/internal/cache/cache_test.go @@ -621,7 +621,8 @@ func (t *inmemTrx) IndexStore() storage.IndexStore { return t.indexStore } func (t *inmemTrx) ChunkStore() storage.ChunkStore { return t.chunkStore } func (t *inmemTrx) Commit() error { return nil } -func (t *inmemStorage) Close() error { return nil } +func (t *inmemStorage) Close() error { return nil } +func (t *inmemStorage) StartSamplingSession() func() { return func() {} } func (t *inmemStorage) Run(ctx context.Context, f func(s transaction.Store) error) error { trx, done := t.NewTransaction(ctx) defer done() diff --git a/pkg/storer/internal/chunkstore/chunkstore.go b/pkg/storer/internal/chunkstore/chunkstore.go index 0b70134f38b..8df430f502f 100644 --- a/pkg/storer/internal/chunkstore/chunkstore.go +++ b/pkg/storer/internal/chunkstore/chunkstore.go @@ -73,6 +73,54 @@ func GetInto(ctx context.Context, r storage.Reader, s storage.Sharky, addr swarm return n, nil } +// LocationToChunkLocation converts internal sharky.Location into storage.ChunkLocation. +func LocationToChunkLocation(l sharky.Location) storage.ChunkLocation { + var cl storage.ChunkLocation + cl[0] = l.Shard + binary.BigEndian.PutUint32(cl[1:5], l.Slot) + binary.BigEndian.PutUint16(cl[5:7], l.Length) + return cl +} + +// ChunkLocationToLocation converts storage.ChunkLocation back into sharky.Location. +func ChunkLocationToLocation(cl storage.ChunkLocation) sharky.Location { + return sharky.Location{ + Shard: cl[0], + Slot: binary.BigEndian.Uint32(cl[1:5]), + Length: binary.BigEndian.Uint16(cl[5:7]), + } +} + +// GetIntoLoc reads chunk data using a storage.ChunkLocation hint. +// If the location is zero or was freed during an active sampling session, or if +// reading sharky directly fails, it falls back to standard index-based GetInto. +func GetIntoLoc( + ctx context.Context, + r storage.Reader, + s storage.Sharky, + guard *LocationGuard, + addr swarm.Address, + loc storage.ChunkLocation, + buf []byte, +) (int, error) { + if loc.IsZero() || (guard != nil && guard.IsFreed(loc)) { + return GetInto(ctx, r, s, addr, buf) + } + + shLoc := ChunkLocationToLocation(loc) + n := int(shLoc.Length) + if len(buf) < n { + return 0, fmt.Errorf("chunk store: buffer too small: %d < %d", len(buf), n) + } + + err := s.Read(ctx, shLoc, buf[:n]) + if err != nil { + // Fallback to verified lookup in case of error + return GetInto(ctx, r, s, addr, buf) + } + 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) @@ -92,6 +140,11 @@ func Has(_ context.Context, r storage.Reader, addr swarm.Address) (bool, error) } func Put(ctx context.Context, s storage.IndexStore, sh storage.Sharky, ch swarm.Chunk) error { + _, err := PutLoc(ctx, s, sh, ch) + return err +} + +func PutLoc(ctx context.Context, s storage.IndexStore, sh storage.Sharky, ch swarm.Chunk) (storage.ChunkLocation, error) { var ( rIdx = &RetrievalIndexItem{Address: ch.Address()} loc sharky.Location @@ -103,44 +156,55 @@ func Put(ctx context.Context, s storage.IndexStore, sh storage.Sharky, ch swarm. // in sharky and create the new indexes. loc, err = sh.Write(ctx, ch.Data()) if err != nil { - return fmt.Errorf("chunk store: write to sharky failed: %w", err) + return storage.ChunkLocation{}, fmt.Errorf("chunk store: write to sharky failed: %w", err) } rIdx.Location = loc rIdx.Timestamp = uint64(time.Now().Unix()) case err != nil: - return fmt.Errorf("chunk store: failed to read: %w", err) + return storage.ChunkLocation{}, fmt.Errorf("chunk store: failed to read: %w", err) + default: + loc = rIdx.Location } rIdx.RefCnt++ - return s.Put(rIdx) + return LocationToChunkLocation(loc), s.Put(rIdx) +} + +func Replace(ctx context.Context, s storage.IndexStore, sh storage.Sharky, guard *LocationGuard, ch swarm.Chunk, emplace bool) error { + _, err := ReplaceLoc(ctx, s, sh, guard, ch, emplace) + return err } -func Replace(ctx context.Context, s storage.IndexStore, sh storage.Sharky, ch swarm.Chunk, emplace bool) error { +func ReplaceLoc(ctx context.Context, s storage.IndexStore, sh storage.Sharky, guard *LocationGuard, ch swarm.Chunk, emplace bool) (storage.ChunkLocation, error) { rIdx := &RetrievalIndexItem{Address: ch.Address()} err := s.Get(rIdx) if err != nil { - return fmt.Errorf("chunk store: failed to read retrievalIndex for address %s: %w", ch.Address(), err) + return storage.ChunkLocation{}, fmt.Errorf("chunk store: failed to read retrievalIndex for address %s: %w", ch.Address(), err) } - err = sh.Release(ctx, rIdx.Location) + oldLoc := rIdx.Location + err = sh.Release(ctx, oldLoc) if err != nil { - return fmt.Errorf("chunkstore: failed to release sharky location: %w", err) + return storage.ChunkLocation{}, fmt.Errorf("chunkstore: failed to release sharky location: %w", err) + } + if guard != nil { + guard.MarkFreed(LocationToChunkLocation(oldLoc)) } loc, err := sh.Write(ctx, ch.Data()) if err != nil { - return fmt.Errorf("chunk store: write to sharky failed: %w", err) + return storage.ChunkLocation{}, fmt.Errorf("chunk store: write to sharky failed: %w", err) } rIdx.Location = loc rIdx.Timestamp = uint64(time.Now().Unix()) if emplace { rIdx.RefCnt++ } - return s.Put(rIdx) + return LocationToChunkLocation(loc), s.Put(rIdx) } -func Delete(ctx context.Context, s storage.IndexStore, sh storage.Sharky, addr swarm.Address) error { +func Delete(ctx context.Context, s storage.IndexStore, sh storage.Sharky, guard *LocationGuard, addr swarm.Address) error { rIdx := &RetrievalIndexItem{Address: addr} err := s.Get(rIdx) switch { @@ -160,6 +224,10 @@ func Delete(ctx context.Context, s storage.IndexStore, sh storage.Sharky, addr s return nil } + if guard != nil { + guard.MarkFreed(LocationToChunkLocation(rIdx.Location)) + } + return errors.Join( sh.Release(ctx, rIdx.Location), s.Delete(rIdx), diff --git a/pkg/storer/internal/chunkstore/chunkstore_test.go b/pkg/storer/internal/chunkstore/chunkstore_test.go index d07728d321d..f1deeba1a76 100644 --- a/pkg/storer/internal/chunkstore/chunkstore_test.go +++ b/pkg/storer/internal/chunkstore/chunkstore_test.go @@ -580,3 +580,80 @@ func BenchmarkChunkStoreGet(b *testing.B) { } }) } + +func TestLocatingChunkStore(t *testing.T) { + t.Parallel() + + fs := afero.NewMemMapFs() + sharky, err := sharky.New(&memFS{Fs: fs}, 1, swarm.SocMaxChunkSize) + if err != nil { + t.Fatal(err) + } + + store := inmemstore.New() + st := transaction.NewStorage(sharky, store) + defer st.Close() + + ch := chunktest.GenerateTestRandomChunk() + ctx := context.Background() + + var loc storage.ChunkLocation + err = st.Run(ctx, func(s transaction.Store) error { + lp, ok := s.ChunkStore().(storage.LocatingPutter) + if !ok { + return errors.New("chunkStore does not implement LocatingPutter") + } + var err error + loc, err = lp.PutLoc(ctx, ch) + return err + }) + if err != nil { + t.Fatalf("putLoc: %v", err) + } + if loc.IsZero() { + t.Fatal("expected non-zero ChunkLocation from PutLoc") + } + + cs := st.ChunkStore() + lg, ok := cs.(storage.LocatingGetterInto) + if !ok { + t.Fatal("chunkStore does not implement LocatingGetterInto") + } + + buf := make([]byte, swarm.SocMaxChunkSize) + + // 1. Direct read via valid ChunkLocation + n, err := lg.GetIntoLoc(ctx, ch.Address(), loc, buf) + if err != nil { + t.Fatalf("getIntoLoc: %v", err) + } + if !bytes.Equal(buf[:n], ch.Data()) { + t.Fatal("chunk data does not match") + } + + // 2. Fallback read via zero ChunkLocation + n, err = lg.GetIntoLoc(ctx, ch.Address(), storage.ChunkLocation{}, buf) + if err != nil { + t.Fatalf("getIntoLoc zero: %v", err) + } + if !bytes.Equal(buf[:n], ch.Data()) { + t.Fatal("chunk data does not match on zero location fallback") + } + + // 3. Guard active and chunk deleted: GetIntoLoc falls back and returns ErrNotFound + done := st.StartSamplingSession() + + err = st.Run(ctx, func(s transaction.Store) error { + return s.ChunkStore().Delete(ctx, ch.Address()) + }) + if err != nil { + t.Fatalf("delete: %v", err) + } + + _, err = lg.GetIntoLoc(ctx, ch.Address(), loc, buf) + if !errors.Is(err, storage.ErrNotFound) { + t.Fatalf("expected ErrNotFound for deleted chunk, got %v", err) + } + + done() +} diff --git a/pkg/storer/internal/chunkstore/guard.go b/pkg/storer/internal/chunkstore/guard.go new file mode 100644 index 00000000000..0b22f63e95b --- /dev/null +++ b/pkg/storer/internal/chunkstore/guard.go @@ -0,0 +1,79 @@ +// Copyright 2024 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package chunkstore + +import ( + "sync" + + "github.com/ethersphere/bee/v2/pkg/storage" +) + +// LocationGuard tracks storage chunk locations that have been released (evicted or replaced) +// during an active sampling session. This protects against use-after-free and silent data +// corruption if Sharky reuses an empty slot while sampling is concurrently reading. +type LocationGuard struct { + mu sync.RWMutex + active int32 + freed map[storage.ChunkLocation]struct{} +} + +// NewLocationGuard returns an initialized LocationGuard. +func NewLocationGuard() *LocationGuard { + return &LocationGuard{} +} + +// StartSession marks a sampling session as active and returns a completion callback. +func (g *LocationGuard) StartSession() func() { + if g == nil { + return func() {} + } + g.mu.Lock() + defer g.mu.Unlock() + + if g.active == 0 { + g.freed = make(map[storage.ChunkLocation]struct{}) + } + g.active++ + + var once sync.Once + return func() { + once.Do(func() { + g.mu.Lock() + defer g.mu.Unlock() + g.active-- + if g.active == 0 { + g.freed = nil + } + }) + } +} + +// MarkFreed records that a ChunkLocation was released. If no session is active, it is a no-op. +func (g *LocationGuard) MarkFreed(loc storage.ChunkLocation) { + if g == nil { + return + } + g.mu.Lock() + defer g.mu.Unlock() + + if g.active > 0 { + g.freed[loc] = struct{}{} + } +} + +// IsFreed returns whether the given ChunkLocation was released during an active session. +func (g *LocationGuard) IsFreed(loc storage.ChunkLocation) bool { + if g == nil { + return false + } + g.mu.RLock() + defer g.mu.RUnlock() + + if g.active == 0 { + return false + } + _, found := g.freed[loc] + return found +} diff --git a/pkg/storer/internal/chunkstore/guard_test.go b/pkg/storer/internal/chunkstore/guard_test.go new file mode 100644 index 00000000000..eed156dcc6e --- /dev/null +++ b/pkg/storer/internal/chunkstore/guard_test.go @@ -0,0 +1,69 @@ +// Copyright 2024 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package chunkstore_test + +import ( + "sync" + "testing" + + "github.com/ethersphere/bee/v2/pkg/storage" + "github.com/ethersphere/bee/v2/pkg/storer/internal/chunkstore" +) + +func TestLocationGuard(t *testing.T) { + t.Parallel() + + guard := chunkstore.NewLocationGuard() + loc1 := storage.ChunkLocation{1, 0, 0, 0, 10, 0, 4} + loc2 := storage.ChunkLocation{1, 0, 0, 0, 20, 0, 4} + + // Inactive guard: MarkFreed should be a no-op + guard.MarkFreed(loc1) + if guard.IsFreed(loc1) { + t.Fatal("expected IsFreed to be false when guard is inactive") + } + + // Start session + done := guard.StartSession() + + // Marking freed during active session + guard.MarkFreed(loc1) + if !guard.IsFreed(loc1) { + t.Fatal("expected IsFreed to be true for loc1") + } + if guard.IsFreed(loc2) { + t.Fatal("expected IsFreed to be false for loc2") + } + + // End session + done() + + if guard.IsFreed(loc1) { + t.Fatal("expected IsFreed to be false after session completed") + } +} + +func TestLocationGuardConcurrency(t *testing.T) { + t.Parallel() + + guard := chunkstore.NewLocationGuard() + done := guard.StartSession() + defer done() + + var wg sync.WaitGroup + for i := range 100 { + wg.Add(2) + loc := storage.ChunkLocation{1, byte(i), 0, 0, 0, 0, 0, 0} + go func() { + defer wg.Done() + guard.MarkFreed(loc) + }() + go func() { + defer wg.Done() + _ = guard.IsFreed(loc) + }() + } + wg.Wait() +} diff --git a/pkg/storer/internal/internal.go b/pkg/storer/internal/internal.go index 9897138d812..0e2d0607a6e 100644 --- a/pkg/storer/internal/internal.go +++ b/pkg/storer/internal/internal.go @@ -77,7 +77,8 @@ func (t *inmemTrx) IndexStore() storage.IndexStore { return t.indexStore } func (t *inmemTrx) ChunkStore() storage.ChunkStore { return t.chunkStore } func (t *inmemTrx) Commit() error { return nil } -func (t *inmemStorage) Close() error { return nil } +func (t *inmemStorage) Close() error { return nil } +func (t *inmemStorage) StartSamplingSession() func() { return func() {} } func (t *inmemStorage) Run(ctx context.Context, f func(s transaction.Store) error) error { trx, done := t.NewTransaction(ctx) defer done() diff --git a/pkg/storer/internal/reserve/items.go b/pkg/storer/internal/reserve/items.go index e05b5c0bed4..d95f6ff33fa 100644 --- a/pkg/storer/internal/reserve/items.go +++ b/pkg/storer/internal/reserve/items.go @@ -112,6 +112,7 @@ type ChunkBinItem struct { BatchID []byte StampHash []byte ChunkType swarm.ChunkType + Location storage.ChunkLocation } func (c *ChunkBinItem) Namespace() string { @@ -144,17 +145,22 @@ func (c *ChunkBinItem) Clone() storage.Item { BatchID: copyBytes(c.BatchID), StampHash: copyBytes(c.StampHash), ChunkType: c.ChunkType, + Location: c.Location, } } -const chunkBinItemSize = 1 + 8 + swarm.HashSize + swarm.HashSize + 1 + swarm.HashSize +const ( + legacyChunkBinItemSize = 1 + 8 + swarm.HashSize + swarm.HashSize + 1 + swarm.HashSize // 106 + chunkBinItemSizeWithLoc = legacyChunkBinItemSize + 8 // 114 + chunkBinItemSize = chunkBinItemSizeWithLoc +) func (c *ChunkBinItem) Marshal() ([]byte, error) { if c.Address.IsZero() { return nil, errMarshalInvalidAddress } - buf := make([]byte, chunkBinItemSize) + buf := make([]byte, chunkBinItemSizeWithLoc) i := 0 buf[i] = c.Bin @@ -173,11 +179,14 @@ func (c *ChunkBinItem) Marshal() ([]byte, error) { i += 1 copy(buf[i:i+swarm.HashSize], c.StampHash) + i += swarm.HashSize + + copy(buf[i:i+8], c.Location[:]) return buf, nil } func (c *ChunkBinItem) Unmarshal(buf []byte) error { - if len(buf) != chunkBinItemSize { + if len(buf) != legacyChunkBinItemSize && len(buf) != chunkBinItemSizeWithLoc { return errUnmarshalInvalidSize } @@ -198,6 +207,13 @@ func (c *ChunkBinItem) Unmarshal(buf []byte) error { i += 1 c.StampHash = copyBytes(buf[i : i+swarm.HashSize]) + i += swarm.HashSize + + if len(buf) == chunkBinItemSizeWithLoc { + copy(c.Location[:], buf[i:i+8]) + } else { + c.Location = storage.ChunkLocation{} + } return nil } diff --git a/pkg/storer/internal/reserve/items_test.go b/pkg/storer/internal/reserve/items_test.go index 19c22224d3a..ef3c3390a23 100644 --- a/pkg/storer/internal/reserve/items_test.go +++ b/pkg/storer/internal/reserve/items_test.go @@ -146,3 +146,56 @@ func TestReserveItems(t *testing.T) { }) } } + +func TestChunkBinItemDualFormat(t *testing.T) { + t.Parallel() + + addr := swarm.NewAddress(storagetest.MaxAddressBytes[:]) + batchID := []byte("01234567890123456789012345678901") + stampHash := []byte("abcdefghijklmnopqrstuvwxyz123456") + loc := storage.ChunkLocation{1, 2, 3, 4, 5, 6, 7, 8} + + item := &reserve.ChunkBinItem{ + Bin: 5, + BinID: 42, + Address: addr, + BatchID: batchID, + StampHash: stampHash, + ChunkType: swarm.ChunkTypeContentAddressed, + Location: loc, + } + + // Marshal produces new format with location + buf, err := item.Marshal() + if err != nil { + t.Fatalf("marshal: %v", err) + } + if len(buf) != 114 { + t.Fatalf("expected 114 bytes, got %d", len(buf)) + } + + // Unmarshal new format preserves location + recovered := new(reserve.ChunkBinItem) + if err := recovered.Unmarshal(buf); err != nil { + t.Fatalf("unmarshal new format: %v", err) + } + if recovered.Location != loc { + t.Fatalf("expected location %v, got %v", loc, recovered.Location) + } + if !recovered.Address.Equal(addr) { + t.Fatalf("expected address %v, got %v", addr, recovered.Address) + } + + // Unmarshal legacy 106-byte format succeeds with zero location + legacyBuf := buf[:106] + legacyRecovered := new(reserve.ChunkBinItem) + if err := legacyRecovered.Unmarshal(legacyBuf); err != nil { + t.Fatalf("unmarshal legacy format: %v", err) + } + if !legacyRecovered.Location.IsZero() { + t.Fatalf("expected zero location for legacy item, got %v", legacyRecovered.Location) + } + if !legacyRecovered.Address.Equal(addr) { + t.Fatalf("expected address %v, got %v", addr, legacyRecovered.Address) + } +} diff --git a/pkg/storer/internal/reserve/reserve.go b/pkg/storer/internal/reserve/reserve.go index 28c28192e96..2c9e049fc9c 100644 --- a/pkg/storer/internal/reserve/reserve.go +++ b/pkg/storer/internal/reserve/reserve.go @@ -170,10 +170,26 @@ func (r *Reserve) Put(ctx context.Context, chunk swarm.Chunk) error { return err } + oldChunkBinItem := &ChunkBinItem{Bin: oldBatchRadiusItem.Bin, BinID: oldBatchRadiusItem.BinID} + _ = s.IndexStore().Get(oldChunkBinItem) + loc := oldChunkBinItem.Location + + if chunkType == swarm.ChunkTypeSingleOwner { + r.logger.Debug("replacing soc in chunkstore", "address", chunk.Address()) + if lr, ok := s.ChunkStore().(storage.LocatingReplacer); ok { + loc, err = lr.ReplaceLoc(ctx, chunk, false) + } else { + err = s.ChunkStore().Replace(ctx, chunk, false) + } + if err != nil { + return err + } + } + // delete old chunk index items err = errors.Join( s.IndexStore().Delete(oldBatchRadiusItem), - s.IndexStore().Delete(&ChunkBinItem{Bin: oldBatchRadiusItem.Bin, BinID: oldBatchRadiusItem.BinID}), + s.IndexStore().Delete(oldChunkBinItem), stampindex.Delete(s.IndexStore(), reserveScope, oldStamp), chunkstamp.DeleteWithStamp(s.IndexStore(), reserveScope, oldBatchRadiusItem.Address, oldStamp), ) @@ -203,17 +219,13 @@ func (r *Reserve) Put(ctx context.Context, chunk swarm.Chunk) error { BatchID: chunk.Stamp().BatchID(), ChunkType: chunkType, StampHash: stampHash, + Location: loc, }), ) if err != nil { return err } - if chunkType == swarm.ChunkTypeSingleOwner { - r.logger.Debug("replacing soc in chunkstore", "address", chunk.Address()) - return s.ChunkStore().Replace(ctx, chunk, false) - } - return nil } @@ -241,6 +253,34 @@ func (r *Reserve) Put(ctx context.Context, chunk swarm.Chunk) error { return err } + var loc storage.ChunkLocation + if chunkType == swarm.ChunkTypeSingleOwner { + var has bool + has, err = s.ChunkStore().Has(ctx, chunk.Address()) + if err != nil { + return err + } + if has { + r.logger.Debug("replacing soc in chunkstore", "address", chunk.Address()) + if lr, ok := s.ChunkStore().(storage.LocatingReplacer); ok { + loc, err = lr.ReplaceLoc(ctx, chunk, true) + } else { + err = s.ChunkStore().Replace(ctx, chunk, true) + } + } else if lp, ok := s.ChunkStore().(storage.LocatingPutter); ok { + loc, err = lp.PutLoc(ctx, chunk) + } else { + err = s.ChunkStore().Put(ctx, chunk) + } + } else if lp, ok := s.ChunkStore().(storage.LocatingPutter); ok { + loc, err = lp.PutLoc(ctx, chunk) + } else { + err = s.ChunkStore().Put(ctx, chunk) + } + if err != nil { + return err + } + err = errors.Join( chunkstamp.Store(s.IndexStore(), reserveScope, chunk), s.IndexStore().Put(&BatchRadiusItem{ @@ -257,36 +297,16 @@ func (r *Reserve) Put(ctx context.Context, chunk swarm.Chunk) error { BatchID: chunk.Stamp().BatchID(), ChunkType: chunkType, StampHash: stampHash, + Location: loc, }), ) if err != nil { return err } - var has bool - if chunkType == swarm.ChunkTypeSingleOwner { - has, err = s.ChunkStore().Has(ctx, chunk.Address()) - if err != nil { - return err - } - if has { - r.logger.Debug("replacing soc in chunkstore", "address", chunk.Address()) - err = s.ChunkStore().Replace(ctx, chunk, true) - } else { - err = s.ChunkStore().Put(ctx, chunk) - } - } else { - err = s.ChunkStore().Put(ctx, chunk) - } - - if err != nil { - return err - } - if !loadedStampIndex { shouldIncReserveSize = true } - return nil }) if err != nil { diff --git a/pkg/storer/internal/reserve/reserve_test.go b/pkg/storer/internal/reserve/reserve_test.go index d67ec34916e..ab9d8914987 100644 --- a/pkg/storer/internal/reserve/reserve_test.go +++ b/pkg/storer/internal/reserve/reserve_test.go @@ -8,8 +8,10 @@ import ( "bytes" "context" "errors" + "io/fs" "math" "math/rand" + "os" "testing" "testing/synctest" @@ -17,8 +19,10 @@ import ( "github.com/ethersphere/bee/v2/pkg/log" "github.com/ethersphere/bee/v2/pkg/postage" postagetesting "github.com/ethersphere/bee/v2/pkg/postage/testing" + "github.com/ethersphere/bee/v2/pkg/sharky" soctesting "github.com/ethersphere/bee/v2/pkg/soc/testing" "github.com/ethersphere/bee/v2/pkg/storage" + "github.com/ethersphere/bee/v2/pkg/storage/inmemstore" chunk "github.com/ethersphere/bee/v2/pkg/storage/testing" "github.com/ethersphere/bee/v2/pkg/storer/internal" "github.com/ethersphere/bee/v2/pkg/storer/internal/chunkstamp" @@ -28,6 +32,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/storer/internal/transaction" "github.com/ethersphere/bee/v2/pkg/swarm" kademlia "github.com/ethersphere/bee/v2/pkg/topology/mock" + "github.com/spf13/afero" "github.com/stretchr/testify/assert" ) @@ -1161,3 +1166,77 @@ func checkChunkInIndexStore(t *testing.T, s storage.Reader, bin uint8, binId uin checkStore(t, s, &reserve.BatchRadiusItem{Bin: bin, BatchID: ch.Stamp().BatchID(), Address: ch.Address(), StampHash: stampHash}, false) checkStore(t, s, &reserve.ChunkBinItem{Bin: bin, BinID: binId, StampHash: stampHash}, false) } + +type memFS struct { + afero.Fs +} + +func (m *memFS) Open(path string) (fs.File, error) { + return m.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o644) +} + +func TestReserveChunkLocation(t *testing.T) { + t.Parallel() + + fs := afero.NewMemMapFs() + sharky, err := sharky.New(&memFS{Fs: fs}, 1, swarm.SocMaxChunkSize) + if err != nil { + t.Fatal(err) + } + + store := inmemstore.New() + st := transaction.NewStorage(sharky, store) + defer st.Close() + + baseAddr := swarm.RandAddress(t) + r, err := reserve.New( + baseAddr, + st, + 0, kademlia.NewTopologyDriver(), + log.Noop, + ) + if err != nil { + t.Fatal(err) + } + + ctx := context.Background() + ch := chunk.GenerateTestRandomChunkAt(t, baseAddr, 0) + + err = r.Put(ctx, ch) + if err != nil { + t.Fatalf("put: %v", err) + } + + var foundLoc storage.ChunkLocation + err = r.IterateChunksItems(0, func(ci *reserve.ChunkBinItem) (bool, error) { + if ci.Address.Equal(ch.Address()) { + foundLoc = ci.Location + return true, nil + } + return false, nil + }) + if err != nil { + t.Fatalf("iterate: %v", err) + } + + if foundLoc.IsZero() { + t.Fatal("expected ChunkBinItem to have non-zero Location after Put") + } + + // Read directly using the location hint + cs := st.ChunkStore() + lg, ok := cs.(storage.LocatingGetterInto) + if !ok { + t.Fatal("ChunkStore does not implement LocatingGetterInto") + } + + buf := make([]byte, swarm.SocMaxChunkSize) + n, err := lg.GetIntoLoc(ctx, ch.Address(), foundLoc, buf) + if err != nil { + t.Fatalf("GetIntoLoc: %v", err) + } + + if !bytes.Equal(buf[:n], ch.Data()) { + t.Fatal("chunk data read via location hint does not match") + } +} diff --git a/pkg/storer/internal/transaction/transaction.go b/pkg/storer/internal/transaction/transaction.go index 2aa1072e624..5ba128adc84 100644 --- a/pkg/storer/internal/transaction/transaction.go +++ b/pkg/storer/internal/transaction/transaction.go @@ -55,6 +55,7 @@ type Storage interface { NewTransaction(context.Context) (Transaction, func()) Run(context.Context, func(Store) error) error Close() error + StartSamplingSession() func() } type store struct { @@ -62,10 +63,15 @@ type store struct { bstore storage.BatchStore metrics metrics chunkLocker *multex.Multex + guard *chunkstore.LocationGuard } func NewStorage(sharky *sharky.Store, bstore storage.BatchStore) Storage { - return &store{sharky, bstore, newMetrics(), multex.New()} + return &store{sharky, bstore, newMetrics(), multex.New(), chunkstore.NewLocationGuard()} +} + +func (s *store) StartSamplingSession() func() { + return s.guard.StartSession() } type transaction struct { @@ -94,7 +100,7 @@ func (s *store) NewTransaction(ctx context.Context) (Transaction, func()) { start: time.Now(), batch: b, indexstore: index, - chunkStore: &chunkStoreTrx{index, sharky, s.chunkLocker, make(map[string]struct{}), s.metrics, false}, + chunkStore: &chunkStoreTrx{index, sharky, s.chunkLocker, make(map[string]struct{}), s.metrics, false, s.guard}, sharkyTrx: sharky, metrics: s.metrics, } @@ -121,7 +127,7 @@ func (s *store) IndexStore() storage.Reader { func (s *store) ChunkStore() storage.ReadOnlyChunkStore { indexStore := &indexTrx{s.bstore, nil, s.metrics} sharyTrx := &sharkyTrx{s.sharky, s.metrics, nil, nil} - return &chunkStoreTrx{indexStore, sharyTrx, s.chunkLocker, nil, s.metrics, true} + return &chunkStoreTrx{indexStore, sharyTrx, s.chunkLocker, nil, s.metrics, true, s.guard} } // Run creates a new transaction and gives the caller access to the transaction @@ -208,6 +214,13 @@ func (t *transaction) ChunkStore() storage.ChunkStore { return t.chunkStore } +var ( + _ storage.ChunkStore = (*chunkStoreTrx)(nil) + _ storage.LocatingPutter = (*chunkStoreTrx)(nil) + _ storage.LocatingReplacer = (*chunkStoreTrx)(nil) + _ storage.LocatingGetterInto = (*chunkStoreTrx)(nil) +) + type chunkStoreTrx struct { indexStore storage.IndexStore sharkyTrx *sharkyTrx @@ -215,6 +228,7 @@ type chunkStoreTrx struct { lockedAddrs map[string]struct{} metrics metrics readOnly bool + guard *chunkstore.LocationGuard } func (c *chunkStoreTrx) Get(ctx context.Context, addr swarm.Address) (ch swarm.Chunk, err error) { @@ -232,6 +246,13 @@ func (c *chunkStoreTrx) GetInto(ctx context.Context, addr swarm.Address, buf []b return chunkstore.GetInto(ctx, c.indexStore, c.sharkyTrx, addr, buf) } +func (c *chunkStoreTrx) GetIntoLoc(ctx context.Context, addr swarm.Address, loc storage.ChunkLocation, buf []byte) (n int, err error) { + defer handleMetric("chunkstore_get", c.metrics)(&err) + unlock := c.lock(addr) + defer unlock() + return chunkstore.GetIntoLoc(ctx, c.indexStore, c.sharkyTrx, c.guard, addr, loc, 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) @@ -246,11 +267,18 @@ func (c *chunkStoreTrx) Put(ctx context.Context, ch swarm.Chunk) (err error) { return chunkstore.Put(ctx, c.indexStore, c.sharkyTrx, ch) } +func (c *chunkStoreTrx) PutLoc(ctx context.Context, ch swarm.Chunk) (loc storage.ChunkLocation, err error) { + defer handleMetric("chunkstore_put", c.metrics)(&err) + unlock := c.lock(ch.Address()) + defer unlock() + return chunkstore.PutLoc(ctx, c.indexStore, c.sharkyTrx, ch) +} + func (c *chunkStoreTrx) Delete(ctx context.Context, addr swarm.Address) (err error) { defer handleMetric("chunkstore_delete", c.metrics)(&err) unlock := c.lock(addr) defer unlock() - return chunkstore.Delete(ctx, c.indexStore, c.sharkyTrx, addr) + return chunkstore.Delete(ctx, c.indexStore, c.sharkyTrx, c.guard, addr) } func (c *chunkStoreTrx) Iterate(ctx context.Context, fn storage.IterateChunkFn) (err error) { @@ -262,7 +290,14 @@ func (c *chunkStoreTrx) Replace(ctx context.Context, ch swarm.Chunk, emplace boo defer handleMetric("chunkstore_replace", c.metrics)(&err) unlock := c.lock(ch.Address()) defer unlock() - return chunkstore.Replace(ctx, c.indexStore, c.sharkyTrx, ch, emplace) + return chunkstore.Replace(ctx, c.indexStore, c.sharkyTrx, c.guard, ch, emplace) +} + +func (c *chunkStoreTrx) ReplaceLoc(ctx context.Context, ch swarm.Chunk, emplace bool) (loc storage.ChunkLocation, err error) { + defer handleMetric("chunkstore_replace", c.metrics)(&err) + unlock := c.lock(ch.Address()) + defer unlock() + return chunkstore.ReplaceLoc(ctx, c.indexStore, c.sharkyTrx, c.guard, ch, emplace) } func (c *chunkStoreTrx) lock(addr swarm.Address) func() { diff --git a/pkg/storer/sample.go b/pkg/storer/sample.go index 66a21032f3c..35e9557d9b8 100644 --- a/pkg/storer/sample.go +++ b/pkg/storer/sample.go @@ -22,6 +22,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/postage" "github.com/ethersphere/bee/v2/pkg/safe" "github.com/ethersphere/bee/v2/pkg/soc" + "github.com/ethersphere/bee/v2/pkg/storage" chunk "github.com/ethersphere/bee/v2/pkg/storage/testing" "github.com/ethersphere/bee/v2/pkg/storer/internal/chunkstamp" "github.com/ethersphere/bee/v2/pkg/storer/internal/reserve" @@ -85,6 +86,8 @@ func (db *DB) ReserveSample( workers := max(4, runtime.NumCPU()) t := time.Now() + defer db.StartSamplingSession()() + defer func() { duration := time.Since(t) err := g.Wait() @@ -160,7 +163,15 @@ func (db *DB) ReserveSample( chunkLoadStart := time.Now() - n, err := chunkStore.GetInto(gCtx, chItem.Address, buf) + var ( + n int + err error + ) + if lg, ok := chunkStore.(storage.LocatingGetterInto); ok { + n, err = lg.GetIntoLoc(gCtx, chItem.Address, chItem.Location, buf) + } else { + n, err = chunkStore.GetInto(gCtx, chItem.Address, buf) + } chunkLoadDuration := time.Since(chunkLoadStart) if err != nil { diff --git a/pkg/storer/storer.go b/pkg/storer/storer.go index 43563683a9a..2563630845a 100644 --- a/pkg/storer/storer.go +++ b/pkg/storer/storer.go @@ -19,14 +19,12 @@ import ( "time" "github.com/ethersphere/bee/v2/pkg/log" - "github.com/ethersphere/bee/v2/pkg/stabilization" - "github.com/ethersphere/bee/v2/pkg/storer/internal/transaction" - m "github.com/ethersphere/bee/v2/pkg/metrics" "github.com/ethersphere/bee/v2/pkg/postage" "github.com/ethersphere/bee/v2/pkg/pusher" "github.com/ethersphere/bee/v2/pkg/retrieval" "github.com/ethersphere/bee/v2/pkg/sharky" + "github.com/ethersphere/bee/v2/pkg/stabilization" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/storage/leveldbstore" "github.com/ethersphere/bee/v2/pkg/storage/migration" @@ -34,6 +32,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/storer/internal/events" pinstore "github.com/ethersphere/bee/v2/pkg/storer/internal/pinning" "github.com/ethersphere/bee/v2/pkg/storer/internal/reserve" + "github.com/ethersphere/bee/v2/pkg/storer/internal/transaction" "github.com/ethersphere/bee/v2/pkg/storer/internal/upload" localmigration "github.com/ethersphere/bee/v2/pkg/storer/migration" "github.com/ethersphere/bee/v2/pkg/swarm" @@ -716,6 +715,13 @@ func (db *DB) Storage() transaction.Storage { return db.storage } +func (db *DB) StartSamplingSession() func() { + if db.storage == nil { + return func() {} + } + return db.storage.StartSamplingSession() +} + type putterSession struct { storage.Putter done func(swarm.Address) error From 07c29026f71d40dd6cc330c40170480650496f27 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Tue, 15 Sep 2026 08:53:27 +0200 Subject: [PATCH 07/10] fix(storer): require active guard session before trusting chunk location hints Ensure GetIntoLoc defaults to fail-safe behavior by falling back to standard verified GetInto whenever no sampling session is active: - Add LocationGuard.SessionActive() method - Check !guard.SessionActive() in GetIntoLoc before using sharky location - Update tests to verify fallback when inactive and direct read when active --- pkg/storer/internal/chunkstore/chunkstore.go | 2 +- pkg/storer/internal/chunkstore/chunkstore_test.go | 15 ++++++++++++--- pkg/storer/internal/chunkstore/guard.go | 11 +++++++++++ pkg/storer/internal/chunkstore/guard_test.go | 9 +++++++++ pkg/storer/internal/reserve/reserve_test.go | 5 ++++- 5 files changed, 37 insertions(+), 5 deletions(-) diff --git a/pkg/storer/internal/chunkstore/chunkstore.go b/pkg/storer/internal/chunkstore/chunkstore.go index 8df430f502f..a12d1742ca6 100644 --- a/pkg/storer/internal/chunkstore/chunkstore.go +++ b/pkg/storer/internal/chunkstore/chunkstore.go @@ -103,7 +103,7 @@ func GetIntoLoc( loc storage.ChunkLocation, buf []byte, ) (int, error) { - if loc.IsZero() || (guard != nil && guard.IsFreed(loc)) { + if loc.IsZero() || guard == nil || !guard.SessionActive() || guard.IsFreed(loc) { return GetInto(ctx, r, s, addr, buf) } diff --git a/pkg/storer/internal/chunkstore/chunkstore_test.go b/pkg/storer/internal/chunkstore/chunkstore_test.go index f1deeba1a76..dcec33d1ce6 100644 --- a/pkg/storer/internal/chunkstore/chunkstore_test.go +++ b/pkg/storer/internal/chunkstore/chunkstore_test.go @@ -622,10 +622,10 @@ func TestLocatingChunkStore(t *testing.T) { buf := make([]byte, swarm.SocMaxChunkSize) - // 1. Direct read via valid ChunkLocation + // 1. Without active session, GetIntoLoc falls back to standard GetInto safely n, err := lg.GetIntoLoc(ctx, ch.Address(), loc, buf) if err != nil { - t.Fatalf("getIntoLoc: %v", err) + t.Fatalf("getIntoLoc without session: %v", err) } if !bytes.Equal(buf[:n], ch.Data()) { t.Fatal("chunk data does not match") @@ -640,9 +640,18 @@ func TestLocatingChunkStore(t *testing.T) { t.Fatal("chunk data does not match on zero location fallback") } - // 3. Guard active and chunk deleted: GetIntoLoc falls back and returns ErrNotFound + // 3. With active session, direct read succeeds done := st.StartSamplingSession() + n, err = lg.GetIntoLoc(ctx, ch.Address(), loc, buf) + if err != nil { + t.Fatalf("getIntoLoc with active session: %v", err) + } + if !bytes.Equal(buf[:n], ch.Data()) { + t.Fatal("chunk data does not match with active session") + } + + // 4. Guard active and chunk deleted: GetIntoLoc falls back and returns ErrNotFound err = st.Run(ctx, func(s transaction.Store) error { return s.ChunkStore().Delete(ctx, ch.Address()) }) diff --git a/pkg/storer/internal/chunkstore/guard.go b/pkg/storer/internal/chunkstore/guard.go index 0b22f63e95b..ead69b268ef 100644 --- a/pkg/storer/internal/chunkstore/guard.go +++ b/pkg/storer/internal/chunkstore/guard.go @@ -77,3 +77,14 @@ func (g *LocationGuard) IsFreed(loc storage.ChunkLocation) bool { _, found := g.freed[loc] return found } + +// SessionActive reports whether at least one sampling session is currently active. +func (g *LocationGuard) SessionActive() bool { + if g == nil { + return false + } + g.mu.RLock() + defer g.mu.RUnlock() + + return g.active > 0 +} diff --git a/pkg/storer/internal/chunkstore/guard_test.go b/pkg/storer/internal/chunkstore/guard_test.go index eed156dcc6e..58d05cedcec 100644 --- a/pkg/storer/internal/chunkstore/guard_test.go +++ b/pkg/storer/internal/chunkstore/guard_test.go @@ -20,6 +20,9 @@ func TestLocationGuard(t *testing.T) { loc2 := storage.ChunkLocation{1, 0, 0, 0, 20, 0, 4} // Inactive guard: MarkFreed should be a no-op + if guard.SessionActive() { + t.Fatal("expected SessionActive to be false initially") + } guard.MarkFreed(loc1) if guard.IsFreed(loc1) { t.Fatal("expected IsFreed to be false when guard is inactive") @@ -27,6 +30,9 @@ func TestLocationGuard(t *testing.T) { // Start session done := guard.StartSession() + if !guard.SessionActive() { + t.Fatal("expected SessionActive to be true during session") + } // Marking freed during active session guard.MarkFreed(loc1) @@ -40,6 +46,9 @@ func TestLocationGuard(t *testing.T) { // End session done() + if guard.SessionActive() { + t.Fatal("expected SessionActive to be false after session completed") + } if guard.IsFreed(loc1) { t.Fatal("expected IsFreed to be false after session completed") } diff --git a/pkg/storer/internal/reserve/reserve_test.go b/pkg/storer/internal/reserve/reserve_test.go index ab9d8914987..f9ed5219b26 100644 --- a/pkg/storer/internal/reserve/reserve_test.go +++ b/pkg/storer/internal/reserve/reserve_test.go @@ -1223,7 +1223,10 @@ func TestReserveChunkLocation(t *testing.T) { t.Fatal("expected ChunkBinItem to have non-zero Location after Put") } - // Read directly using the location hint + // Read directly using the location hint within a sampling session + done := st.StartSamplingSession() + defer done() + cs := st.ChunkStore() lg, ok := cs.(storage.LocatingGetterInto) if !ok { From 8011c44377fbb909bc32e43b5bbe3bbe3eabd4e2 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Tue, 15 Sep 2026 09:11:32 +0200 Subject: [PATCH 08/10] fix(storer): mark location as freed before release in ReplaceLoc Align execution order in ReplaceLoc with Delete to enforce the principle of marking resources in LocationGuard prior to releasing them in Sharky: - Move guard.MarkFreed before sh.Release in ReplaceLoc - Add test coverage verifying that old locations trigger safe indexStore fallback after ReplaceLoc --- pkg/storer/internal/chunkstore/chunkstore.go | 7 +-- .../internal/chunkstore/chunkstore_test.go | 43 +++++++++++++++++++ 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/pkg/storer/internal/chunkstore/chunkstore.go b/pkg/storer/internal/chunkstore/chunkstore.go index a12d1742ca6..c698dfb72f1 100644 --- a/pkg/storer/internal/chunkstore/chunkstore.go +++ b/pkg/storer/internal/chunkstore/chunkstore.go @@ -184,13 +184,14 @@ func ReplaceLoc(ctx context.Context, s storage.IndexStore, sh storage.Sharky, gu } oldLoc := rIdx.Location + if guard != nil { + guard.MarkFreed(LocationToChunkLocation(oldLoc)) + } + err = sh.Release(ctx, oldLoc) if err != nil { return storage.ChunkLocation{}, fmt.Errorf("chunkstore: failed to release sharky location: %w", err) } - if guard != nil { - guard.MarkFreed(LocationToChunkLocation(oldLoc)) - } loc, err := sh.Write(ctx, ch.Data()) if err != nil { diff --git a/pkg/storer/internal/chunkstore/chunkstore_test.go b/pkg/storer/internal/chunkstore/chunkstore_test.go index dcec33d1ce6..f46d3784c7d 100644 --- a/pkg/storer/internal/chunkstore/chunkstore_test.go +++ b/pkg/storer/internal/chunkstore/chunkstore_test.go @@ -664,5 +664,48 @@ func TestLocatingChunkStore(t *testing.T) { t.Fatalf("expected ErrNotFound for deleted chunk, got %v", err) } + // 5. ReplaceLoc marks old location as freed, while new location is valid + ch2 := chunktest.GenerateTestRandomChunk() + var loc2 storage.ChunkLocation + err = st.Run(ctx, func(s transaction.Store) error { + lp := s.ChunkStore().(storage.LocatingPutter) + loc2, err = lp.PutLoc(ctx, ch2) + return err + }) + if err != nil { + t.Fatalf("putLoc ch2: %v", err) + } + + updatedData := append(ch2.Data()[:swarm.SpanSize], bytes.Repeat([]byte{0x42}, len(ch2.Data())-swarm.SpanSize)...) + ch2Updated := swarm.NewChunk(ch2.Address(), updatedData) + + var newLoc2 storage.ChunkLocation + err = st.Run(ctx, func(s transaction.Store) error { + lr := s.ChunkStore().(storage.LocatingReplacer) + newLoc2, err = lr.ReplaceLoc(ctx, ch2Updated, false) + return err + }) + if err != nil { + t.Fatalf("replaceLoc ch2: %v", err) + } + + // Reading with old location hint must fallback to indexStore and return updated data + n, err = lg.GetIntoLoc(ctx, ch2.Address(), loc2, buf) + if err != nil { + t.Fatalf("getIntoLoc old loc2: %v", err) + } + if !bytes.Equal(buf[:n], ch2Updated.Data()) { + t.Fatal("expected updated chunk data on old location fallback") + } + + // Reading with new location hint reads directly and matches updated data + n, err = lg.GetIntoLoc(ctx, ch2.Address(), newLoc2, buf) + if err != nil { + t.Fatalf("getIntoLoc new loc2: %v", err) + } + if !bytes.Equal(buf[:n], ch2Updated.Data()) { + t.Fatal("expected updated chunk data on new location") + } + done() } From afd6fdb81f5dd6e5d9fb079f71e9381e40fc335f Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Tue, 15 Sep 2026 09:31:34 +0200 Subject: [PATCH 09/10] test(storer): document 4-part guard invariant chain and verify GetIntoLoc address locking --- pkg/storer/internal/chunkstore/guard.go | 25 +++ .../internal/transaction/transaction.go | 8 + .../internal/transaction/transaction_test.go | 158 ++++++++++++++++++ 3 files changed, 191 insertions(+) diff --git a/pkg/storer/internal/chunkstore/guard.go b/pkg/storer/internal/chunkstore/guard.go index ead69b268ef..ac1256679c1 100644 --- a/pkg/storer/internal/chunkstore/guard.go +++ b/pkg/storer/internal/chunkstore/guard.go @@ -13,6 +13,31 @@ import ( // LocationGuard tracks storage chunk locations that have been released (evicted or replaced) // during an active sampling session. This protects against use-after-free and silent data // corruption if Sharky reuses an empty slot while sampling is concurrently reading. +// +// Invariant Chain for Safe ChunkLocation Hint Retrieval: +// Direct chunk data retrieval via ChunkLocation hints (bypassing LevelDB index lookups) +// is safe against use-after-free races and Sharky slot reuse IF AND ONLY IF the following +// four invariants hold simultaneously: +// +// 1. Snapshot semantics of iterator (Phase 1): +// When reserve sampling (or any other batch reader) iterates chunk items in Phase 1, +// each ChunkBinItem's ChunkLocation was valid and committed at the moment of iteration. +// +// 2. Per-address mutual exclusion (c.lock(addr) in GetIntoLoc, Put, ReplaceLoc, Delete): +// In chunkStoreTrx, all read and write operations for a given chunk address are serialized +// using the global address locker (Multex). GetIntoLoc MUST acquire c.lock(addr) before +// inspecting the guard or reading from Sharky. +// +// 3. MarkFreed under the same lock prior to slot release: +// In Delete and ReplaceLoc, guard.MarkFreed(loc) is called while holding c.lock(addr) +// BEFORE Sharky.Release(loc) is executed. Because the address lock is held throughout the +// invalidation, any concurrent GetIntoLoc call for that address is blocked. Once unblocked, +// GetIntoLoc will observe guard.IsFreed(loc) == true and will not read the released slot. +// +// 4. Conservative guard with fail-safe fallback: +// If loc.IsZero(), guard == nil, !guard.SessionActive(), guard.IsFreed(loc), or if Sharky +// direct read returns an error, GetIntoLoc immediately and transparently falls back to +// the authoritative, index-backed GetInto(addr) retrieval. type LocationGuard struct { mu sync.RWMutex active int32 diff --git a/pkg/storer/internal/transaction/transaction.go b/pkg/storer/internal/transaction/transaction.go index 5ba128adc84..805be2dcad0 100644 --- a/pkg/storer/internal/transaction/transaction.go +++ b/pkg/storer/internal/transaction/transaction.go @@ -246,6 +246,14 @@ func (c *chunkStoreTrx) GetInto(ctx context.Context, addr swarm.Address, buf []b return chunkstore.GetInto(ctx, c.indexStore, c.sharkyTrx, addr, buf) } +// GetIntoLoc reads chunk data using an opaque ChunkLocation hint. +// +// CRITICAL SAFETY INVARIANT (Pillar 2 of LocationGuard safety chain): +// unlock := c.lock(addr) MUST be held here to guarantee mutual exclusion with concurrent +// Delete and ReplaceLoc operations. Those operations mark locations as freed in LocationGuard +// under the same per-address lock before releasing the Sharky slot. Without this lock, a +// concurrent GetIntoLoc call could read from Sharky after slot release but before or during +// guard update, leading to use-after-free corruption if the slot is reused. func (c *chunkStoreTrx) GetIntoLoc(ctx context.Context, addr swarm.Address, loc storage.ChunkLocation, buf []byte) (n int, err error) { defer handleMetric("chunkstore_get", c.metrics)(&err) unlock := c.lock(addr) diff --git a/pkg/storer/internal/transaction/transaction_test.go b/pkg/storer/internal/transaction/transaction_test.go index 0fcfa42b5ef..398bb721dd9 100644 --- a/pkg/storer/internal/transaction/transaction_test.go +++ b/pkg/storer/internal/transaction/transaction_test.go @@ -5,11 +5,14 @@ package transaction_test import ( + "bytes" "context" + "errors" "io/fs" "os" "path/filepath" "testing" + "time" "github.com/ethersphere/bee/v2/pkg/sharky" "github.com/ethersphere/bee/v2/pkg/storage" @@ -187,3 +190,158 @@ func Test_TransactionStorage(t *testing.T) { } }) } + +func TestGetIntoLoc_LocksAddress(t *testing.T) { + t.Parallel() + + sharkyStore, err := sharky.New(&dirFS{basedir: t.TempDir()}, 32, swarm.SocMaxChunkSize) + assert.NoError(t, err) + + store, _, err := leveldbstore.New("", nil) + assert.NoError(t, err) + + st := transaction.NewStorage(sharkyStore, store) + t.Cleanup(func() { + assert.NoError(t, st.Close()) + }) + + ctx := context.Background() + ch := test.GenerateTestRandomChunk() + + var loc storage.ChunkLocation + err = st.Run(ctx, func(s transaction.Store) error { + lp, ok := s.ChunkStore().(storage.LocatingPutter) + assert.True(t, ok) + var putErr error + loc, putErr = lp.PutLoc(ctx, ch) + return putErr + }) + assert.NoError(t, err) + + sessionDone := st.StartSamplingSession() + defer sessionDone() + + // Start a transaction that locks ch.Address() + tx, txDone := st.NewTransaction(ctx) + defer txDone() + + err = tx.ChunkStore().Delete(ctx, ch.Address()) + assert.NoError(t, err) + + // Call GetIntoLoc in a goroutine; it must block on c.lock(addr) until the transaction finishes. + started := make(chan struct{}) + completed := make(chan struct{}) + lg, ok := st.ChunkStore().(storage.LocatingGetterInto) + assert.True(t, ok) + + buf := make([]byte, swarm.SocMaxChunkSize) + var ( + readErr error + n int + ) + + go func() { + close(started) + n, readErr = lg.GetIntoLoc(ctx, ch.Address(), loc, buf) + close(completed) + }() + + <-started + + // Assert that GetIntoLoc is blocked while tx holds the address lock + select { + case <-completed: + t.Fatal("GetIntoLoc must block while address lock is held by active transaction") + case <-time.After(100 * time.Millisecond): + } + + // Release the address lock by finishing the transaction + txDone() + + // Assert that GetIntoLoc now unblocks and succeeds + select { + case <-completed: + if readErr != nil { + t.Fatalf("expected no error, got %v", readErr) + } + if !bytes.Equal(ch.Data(), buf[:n]) { + t.Fatal("chunk data mismatch") + } + case <-time.After(time.Second): + t.Fatal("GetIntoLoc failed to unblock after transaction finished") + } +} + +func TestGetIntoLoc_ConcurrentDelete_UnderLock(t *testing.T) { + t.Parallel() + + sharkyStore, err := sharky.New(&dirFS{basedir: t.TempDir()}, 32, swarm.SocMaxChunkSize) + assert.NoError(t, err) + + store, _, err := leveldbstore.New("", nil) + assert.NoError(t, err) + + st := transaction.NewStorage(sharkyStore, store) + t.Cleanup(func() { + assert.NoError(t, st.Close()) + }) + + ctx := context.Background() + ch := test.GenerateTestRandomChunk() + + var loc storage.ChunkLocation + err = st.Run(ctx, func(s transaction.Store) error { + lp, ok := s.ChunkStore().(storage.LocatingPutter) + assert.True(t, ok) + var putErr error + loc, putErr = lp.PutLoc(ctx, ch) + return putErr + }) + assert.NoError(t, err) + + sessionDone := st.StartSamplingSession() + defer sessionDone() + + // Start a transaction that deletes ch.Address() + tx, txDone := st.NewTransaction(ctx) + defer txDone() + + // Delete calls guard.MarkFreed(loc) under the address lock + err = tx.ChunkStore().Delete(ctx, ch.Address()) + assert.NoError(t, err) + + started := make(chan struct{}) + completed := make(chan struct{}) + lg, ok := st.ChunkStore().(storage.LocatingGetterInto) + assert.True(t, ok) + + buf := make([]byte, swarm.SocMaxChunkSize) + var readErr error + + go func() { + close(started) + _, readErr = lg.GetIntoLoc(ctx, ch.Address(), loc, buf) + close(completed) + }() + + <-started + + select { + case <-completed: + t.Fatal("GetIntoLoc must block while address lock is held by active transaction") + case <-time.After(100 * time.Millisecond): + } + + // Commit the delete: batch is written, sharky slot is released, address is unlocked + assert.NoError(t, tx.Commit()) + + // Once unblocked, GetIntoLoc sees guard.IsFreed(loc) == true and falls back to index GetInto + select { + case <-completed: + if !errors.Is(readErr, storage.ErrNotFound) { + t.Fatalf("expected ErrNotFound, got %v", readErr) + } + case <-time.After(time.Second): + t.Fatal("GetIntoLoc failed to unblock after commit") + } +} From 034f24215f09ea695f3e44071b1d537c65928897 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Tue, 15 Sep 2026 12:42:46 +0200 Subject: [PATCH 10/10] feat(storer): backfill ChunkBinItem.Location for reserves written before the field The location hint added to ChunkBinItem only lands on entries that are written after the upgrade. Everything already in the reserve stays in the 106-byte legacy layout, Unmarshal leaves its Location zero, and GetIntoLoc falls back to the retrieval-index lookup for every one of those chunks. The dual-format Unmarshal makes the old records readable; nothing populates them, so on an upgraded node the hint is inert until the reserve turns over on its own, which takes days or never. Measured on a 2.41M chunk testnet reserve: GetIntoLoc fell back on 100% of chunks and leveldbstore.Get stayed at 2731 B/chunk, unchanged from the branch without the hint. Add migration step 8, which walks the bin index, resolves each address in the retrieval index once, and writes the location back. Entries are flushed in windows rather than collected up front: the reserve holds millions of items and ReserveRepairer's collect-everything approach would cost hundreds of megabytes while the node is still starting. Since Location lives in the value and the key stays (Bin, BinID), rewriting does not disturb the iteration it runs inside. Each window is spread over NumCPU workers because the per-entry retrieval-index lookup is a random read and would otherwise leave the disk idle. The step is idempotent. Entries that already carry a location are skipped, and entries whose chunk is gone from the chunkstore keep a zero location so the sampler keeps falling back for them. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018s5GBDGHX6Nop4RxHhsCr2 --- pkg/storer/migration/all_steps.go | 1 + pkg/storer/migration/chunkBinItemLocation.go | 142 +++++++++++++ .../migration/chunkBinItemLocation_test.go | 196 ++++++++++++++++++ 3 files changed, 339 insertions(+) create mode 100644 pkg/storer/migration/chunkBinItemLocation.go create mode 100644 pkg/storer/migration/chunkBinItemLocation_test.go diff --git a/pkg/storer/migration/all_steps.go b/pkg/storer/migration/all_steps.go index 118b6ae0d12..d8876642683 100644 --- a/pkg/storer/migration/all_steps.go +++ b/pkg/storer/migration/all_steps.go @@ -31,6 +31,7 @@ func AfterInitSteps( 5: legacyNoopStep, 6: legacyNoopStep, 7: legacyNoopStep, + 8: BackfillChunkBinItemLocation(st, logger), } } diff --git a/pkg/storer/migration/chunkBinItemLocation.go b/pkg/storer/migration/chunkBinItemLocation.go new file mode 100644 index 00000000000..5199060f591 --- /dev/null +++ b/pkg/storer/migration/chunkBinItemLocation.go @@ -0,0 +1,142 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package migration + +import ( + "context" + "errors" + "runtime" + "sync/atomic" + "time" + + "github.com/ethersphere/bee/v2/pkg/log" + "github.com/ethersphere/bee/v2/pkg/storage" + "github.com/ethersphere/bee/v2/pkg/storer/internal/chunkstore" + "github.com/ethersphere/bee/v2/pkg/storer/internal/reserve" + "github.com/ethersphere/bee/v2/pkg/storer/internal/transaction" + "golang.org/x/sync/errgroup" +) + +const ( + // backfillFlushSize is the number of entries held in memory before they are + // written back. The reserve holds millions of bin items, so collecting them + // all the way ReserveRepairer does would cost hundreds of megabytes on a + // node that is still starting up. + backfillFlushSize = 10_000 + // backfillTxSize is how many entries share one transaction inside a window. + backfillTxSize = 500 +) + +// BackfillChunkBinItemLocation fills in ChunkBinItem.Location for entries that +// were written before the field existed. +// +// The sampler reads chunk data straight from sharky when a bin item carries a +// location, and falls back to a retrieval-index lookup when it does not. Bin +// items only gain a location when they are (re)written, so without this step an +// upgraded node keeps paying for the lookup on every chunk already in its +// reserve, and the hint is worthless until the reserve turns over on its own. +// +// Rewriting a bin item changes only the value: the key is (Bin, BinID), so the +// iteration this runs inside is unaffected and the entries can be flushed in +// windows rather than accumulated. +// +// The step is idempotent. Entries that already carry a location are skipped, and +// so are entries whose chunk is no longer in the chunkstore — those keep a zero +// location and the sampler keeps falling back for them, which is correct. +func BackfillChunkBinItemLocation(st transaction.Storage, logger log.Logger) func() error { + return func() error { + start := time.Now() + + var ( + seen int + filled atomic.Int64 + missing atomic.Int64 + window []*reserve.ChunkBinItem + ) + + // Every entry costs a random read of the retrieval index, which is the + // slow part: single file order here would leave the disk idle most of + // the time, so each window is spread over the same number of workers + // ReserveRepairer uses. + flush := func() error { + if len(window) == 0 { + return nil + } + + var eg errgroup.Group + eg.SetLimit(runtime.NumCPU()) + + for i := 0; i < len(window); i += backfillTxSize { + batch := window[i:min(i+backfillTxSize, len(window))] + eg.Go(func() error { + return st.Run(context.Background(), func(s transaction.Store) error { + for _, item := range batch { + rIdx := &chunkstore.RetrievalIndexItem{Address: item.Address} + if err := s.IndexStore().Get(rIdx); err != nil { + if errors.Is(err, storage.ErrNotFound) { + missing.Add(1) + continue + } + return err + } + + item.Location = chunkstore.LocationToChunkLocation(rIdx.Location) + if err := s.IndexStore().Put(item); err != nil { + return err + } + filled.Add(1) + } + return nil + }) + }) + } + + err := eg.Wait() + window = window[:0] + return err + } + + err := st.IndexStore().Iterate( + storage.Query{ + Factory: func() storage.Item { return new(reserve.ChunkBinItem) }, + }, + func(res storage.Result) (bool, error) { + item := res.Entry.(*reserve.ChunkBinItem) + seen++ + if !item.Location.IsZero() { + return false, nil + } + + window = append(window, item) + if len(window) < backfillFlushSize { + return false, nil + } + + if err := flush(); err != nil { + return true, err + } + logger.Info("backfilling chunk bin item locations", "seen", seen, "filled", filled.Load()) + return false, nil + }, + ) + if err != nil { + return err + } + + if err := flush(); err != nil { + return err + } + + logger.Info( + "chunk bin item locations backfilled", + "seen", seen, + "filled", filled.Load(), + "missing_chunks", missing.Load(), + "duration", time.Since(start), + ) + + return nil + } +} diff --git a/pkg/storer/migration/chunkBinItemLocation_test.go b/pkg/storer/migration/chunkBinItemLocation_test.go new file mode 100644 index 00000000000..bcc8331683f --- /dev/null +++ b/pkg/storer/migration/chunkBinItemLocation_test.go @@ -0,0 +1,196 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package migration_test + +import ( + "context" + "io/fs" + "os" + "path/filepath" + "testing" + + "github.com/ethersphere/bee/v2/pkg/log" + "github.com/ethersphere/bee/v2/pkg/sharky" + "github.com/ethersphere/bee/v2/pkg/storage" + "github.com/ethersphere/bee/v2/pkg/storage/inmemstore" + 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/reserve" + "github.com/ethersphere/bee/v2/pkg/storer/internal/transaction" + localmigration "github.com/ethersphere/bee/v2/pkg/storer/migration" + "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/stretchr/testify/assert" +) + +type dirFS struct { + basedir string +} + +func (d *dirFS) Open(path string) (fs.File, error) { + return os.OpenFile(filepath.Join(d.basedir, path), os.O_RDWR|os.O_CREATE, 0o644) +} + +// locatingStorage builds a storage backed by a real sharky store, so that +// ChunkStore().Put writes a retrieval index entry the backfill can read. The +// inmem storage used by the other migration tests keeps chunks in a plain map +// and writes no retrieval index at all. +func locatingStorage(t *testing.T) transaction.Storage { + t.Helper() + + sharkyStore, err := sharky.New(&dirFS{basedir: t.TempDir()}, 1, swarm.SocMaxChunkSize) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := sharkyStore.Close(); err != nil { + t.Errorf("close sharky: %v", err) + } + }) + + return transaction.NewStorage(sharkyStore, inmemstore.New()) +} + +func TestBackfillChunkBinItemLocation(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := locatingStorage(t) + + // chunks that exist in the chunkstore, with bin items carrying no location + stored := make([]swarm.Chunk, 5) + for i := range stored { + ch := chunktest.GenerateTestRandomChunk() + stored[i] = ch + + err := st.Run(ctx, func(s transaction.Store) error { + if err := s.ChunkStore().Put(ctx, ch); err != nil { + return err + } + return s.IndexStore().Put(&reserve.ChunkBinItem{ + Bin: 0, + BinID: uint64(i + 1), + Address: ch.Address(), + ChunkType: swarm.ChunkTypeContentAddressed, + }) + }) + assert.NoError(t, err) + } + + // a bin item whose chunk is not in the chunkstore + orphan := chunktest.GenerateTestRandomChunk() + err := st.Run(ctx, func(s transaction.Store) error { + return s.IndexStore().Put(&reserve.ChunkBinItem{ + Bin: 1, + BinID: 1, + Address: orphan.Address(), + ChunkType: swarm.ChunkTypeContentAddressed, + }) + }) + assert.NoError(t, err) + + // a bin item that already carries a location must be left alone + preset := storage.ChunkLocation{9, 9, 9, 9, 9, 9, 9, 9} + presetCh := chunktest.GenerateTestRandomChunk() + err = st.Run(ctx, func(s transaction.Store) error { + if err := s.ChunkStore().Put(ctx, presetCh); err != nil { + return err + } + return s.IndexStore().Put(&reserve.ChunkBinItem{ + Bin: 2, + BinID: 1, + Address: presetCh.Address(), + ChunkType: swarm.ChunkTypeContentAddressed, + Location: preset, + }) + }) + assert.NoError(t, err) + + assert.NoError(t, localmigration.BackfillChunkBinItemLocation(st, log.Noop)()) + + t.Run("stored chunks get the retrieval index location", func(t *testing.T) { + for i, ch := range stored { + rIdx := &chunkstore.RetrievalIndexItem{Address: ch.Address()} + assert.NoError(t, st.IndexStore().Get(rIdx)) + + item := &reserve.ChunkBinItem{Bin: 0, BinID: uint64(i + 1)} + assert.NoError(t, st.IndexStore().Get(item)) + + assert.False(t, item.Location.IsZero(), "location not filled for chunk %d", i) + assert.Equal(t, chunkstore.LocationToChunkLocation(rIdx.Location), item.Location) + } + }) + + t.Run("missing chunk keeps a zero location", func(t *testing.T) { + item := &reserve.ChunkBinItem{Bin: 1, BinID: 1} + assert.NoError(t, st.IndexStore().Get(item)) + assert.True(t, item.Location.IsZero()) + }) + + t.Run("existing location is preserved", func(t *testing.T) { + item := &reserve.ChunkBinItem{Bin: 2, BinID: 1} + assert.NoError(t, st.IndexStore().Get(item)) + assert.Equal(t, preset, item.Location) + }) + + t.Run("idempotent", func(t *testing.T) { + before := make([]storage.ChunkLocation, len(stored)) + for i := range stored { + item := &reserve.ChunkBinItem{Bin: 0, BinID: uint64(i + 1)} + assert.NoError(t, st.IndexStore().Get(item)) + before[i] = item.Location + } + + assert.NoError(t, localmigration.BackfillChunkBinItemLocation(st, log.Noop)()) + + for i := range stored { + item := &reserve.ChunkBinItem{Bin: 0, BinID: uint64(i + 1)} + assert.NoError(t, st.IndexStore().Get(item)) + assert.Equal(t, before[i], item.Location) + } + }) +} + +// TestBackfillChunkBinItemLocationSpansWindows drives more entries than the +// flush window so the windowed write-back path is exercised, not just a single +// trailing flush. +func TestBackfillChunkBinItemLocationSpansWindows(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := locatingStorage(t) + + const count = 25 + + for i := range count { + ch := chunktest.GenerateTestRandomChunk() + err := st.Run(ctx, func(s transaction.Store) error { + if err := s.ChunkStore().Put(ctx, ch); err != nil { + return err + } + return s.IndexStore().Put(&reserve.ChunkBinItem{ + Bin: 0, + BinID: uint64(i + 1), + Address: ch.Address(), + ChunkType: swarm.ChunkTypeContentAddressed, + }) + }) + assert.NoError(t, err) + } + + assert.NoError(t, localmigration.BackfillChunkBinItemLocation(st, log.Noop)()) + + filled := 0 + err := st.IndexStore().Iterate( + storage.Query{Factory: func() storage.Item { return new(reserve.ChunkBinItem) }}, + func(res storage.Result) (bool, error) { + if !res.Entry.(*reserve.ChunkBinItem).Location.IsZero() { + filled++ + } + return false, nil + }, + ) + assert.NoError(t, err) + assert.Equal(t, count, filled) +}