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/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..97d20e649ed 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. @@ -45,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 @@ -73,6 +104,7 @@ type ChunkGetterDeleter interface { type ChunkStore interface { Getter + GetterInto Putter Deleter Hasser @@ -84,5 +116,6 @@ type ChunkStore interface { type ReadOnlyChunkStore interface { Getter + GetterInto Hasser } 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/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/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/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 6d2745cc5fd..c698dfb72f1 100644 --- a/pkg/storer/internal/chunkstore/chunkstore.go +++ b/pkg/storer/internal/chunkstore/chunkstore.go @@ -51,6 +51,76 @@ 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 +} + +// 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.SessionActive() || 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) @@ -70,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 @@ -81,44 +156,56 @@ 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 + if guard != nil { + guard.MarkFreed(LocationToChunkLocation(oldLoc)) + } + + 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) } 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 { @@ -138,6 +225,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 970e92df9b8..f46d3784c7d 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" @@ -497,17 +496,216 @@ type chunkStore struct { sharky *sharky.Store } -func makeStorage(t *testing.T) *chunkStore { - t.Helper() +func makeStorage(tb testing.TB) *chunkStore { + tb.Helper() store := inmemstore.New() sharky, err := sharky.New(&memFS{Fs: afero.NewMemMapFs()}, 1, swarm.SocMaxChunkSize) - assert.NoError(t, err) + assert.NoError(tb, err) - t.Cleanup(func() { - assert.NoError(t, store.Close()) - assert.NoError(t, sharky.Close()) + tb.Cleanup(func() { + assert.NoError(tb, store.Close()) + assert.NoError(tb, sharky.Close()) }) return &chunkStore{transaction.NewStorage(sharky, store), sharky} } + +// BenchmarkChunkStoreGet measures a single chunk read: one retrieval-index +// lookup followed by one sharky read. It is the micro-benchmark for the read +// path that the reserve sampler drives once per chunk. +// +// The two variants differ only in where the ChunkStore handle comes from. +// "per_call" mirrors what the sampler does today, building a fresh handle for +// every chunk; "hoisted" builds it once. The gap between them is the cost of +// that handle alone. +func BenchmarkChunkStoreGet(b *testing.B) { + ctx := context.Background() + + setup := func(b *testing.B) (*chunkStore, swarm.Address) { + b.Helper() + + st := makeStorage(b) + ch := chunktest.GenerateTestRandomChunk() + + err := st.Run(ctx, func(s transaction.Store) error { + return s.ChunkStore().Put(ctx, ch) + }) + if err != nil { + b.Fatal(err) + } + + return st, ch.Address() + } + + b.Run("per_call", func(b *testing.B) { + st, addr := setup(b) + + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + if _, err := st.ChunkStore().Get(ctx, addr); err != nil { + b.Fatal(err) + } + } + }) + + b.Run("hoisted", func(b *testing.B) { + st, addr := setup(b) + cs := st.ChunkStore() + + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + if _, err := cs.Get(ctx, addr); err != nil { + b.Fatal(err) + } + } + }) + + b.Run("get_into", func(b *testing.B) { + st, addr := setup(b) + cs := st.ChunkStore() + buf := make([]byte, swarm.SocMaxChunkSize) + + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + if _, err := cs.GetInto(ctx, addr, buf); err != nil { + b.Fatal(err) + } + } + }) +} + +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. 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 without session: %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. 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()) + }) + 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) + } + + // 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() +} diff --git a/pkg/storer/internal/chunkstore/guard.go b/pkg/storer/internal/chunkstore/guard.go new file mode 100644 index 00000000000..ac1256679c1 --- /dev/null +++ b/pkg/storer/internal/chunkstore/guard.go @@ -0,0 +1,115 @@ +// 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. +// +// 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 + 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 +} + +// 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 new file mode 100644 index 00000000000..58d05cedcec --- /dev/null +++ b/pkg/storer/internal/chunkstore/guard_test.go @@ -0,0 +1,78 @@ +// 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 + 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") + } + + // 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) + 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.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") + } +} + +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..f9ed5219b26 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,80 @@ 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 within a sampling session + done := st.StartSamplingSession() + defer done() + + 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 0f403580152..805be2dcad0 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) { @@ -225,6 +239,28 @@ 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) +} + +// 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) + 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) @@ -239,11 +275,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) { @@ -255,7 +298,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/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") + } +} 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) +} diff --git a/pkg/storer/sample.go b/pkg/storer/sample.go index f4b6f8139e3..35e9557d9b8 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" @@ -21,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" @@ -37,6 +39,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 @@ -65,7 +73,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{} @@ -78,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() @@ -111,15 +121,15 @@ 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 })) // 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) @@ -127,6 +137,12 @@ 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() + buf := make([]byte, swarm.SocMaxChunkSize) defer func() { addStats(wstat) }() @@ -147,7 +163,15 @@ func (db *DB) ReserveSample( chunkLoadStart := time.Now() - chunk, err := db.ChunkStore().Get(ctx, chItem.Address) + 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 { @@ -159,21 +183,20 @@ 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 } wstat.TaddrDuration += time.Since(taddrStart) select { - case sampleItemChan <- SampleItem{ - TransformedAddress: taddr, - ChunkAddress: chunk.Address(), - ChunkData: chunk.Data(), - Stamp: postage.NewStamp(chItem.BatchID, nil, nil, nil), + case sampleItemChan <- sampleItemInternal{ + transformedAddress: taddr, + chunkAddress: chItem.Address, + batchID: chItem.BatchID, }: - case <-ctx.Done(): - return ctx.Err() + case <-gCtx.Done(): + return gCtx.Err() } } @@ -219,6 +242,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 @@ -226,15 +250,22 @@ 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) + if err != nil { + stats.ChunkLoadFailed++ + db.logger.Debug("failed loading chunk", "chunk_address", item.chunkAddress, "error", err) continue } - ch := swarm.NewChunk(item.ChunkAddress, item.ChunkData).WithStamp(stamp) + 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 { @@ -252,9 +283,12 @@ func (db *DB) ReserveSample( stampValidDuration := time.Since(stampValidStart) stats.ValidStampDuration += stampValidDuration - item.Stamp = postage.NewStamp(stamp.BatchID(), stamp.Index(), stamp.Timestamp(), stamp.Sig()) - - 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++ } } @@ -295,22 +329,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 } @@ -318,20 +355,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 { @@ -400,7 +437,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 16b65d81a09..31f44364ed6 100644 --- a/pkg/storer/sample_test.go +++ b/pkg/storer/sample_test.go @@ -14,8 +14,8 @@ import ( "github.com/ethersphere/bee/v2/pkg/bmt" "github.com/ethersphere/bee/v2/pkg/cac" "github.com/ethersphere/bee/v2/pkg/postage" - 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" @@ -302,6 +302,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 +313,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 +461,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 +507,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 { @@ -458,6 +517,53 @@ func BenchmarkReserveSample1k(b *testing.B) { } } +// 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) { 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