From e737d63ba8fa19bb154068b80dd93a2cc61d58a2 Mon Sep 17 00:00:00 2001 From: Alok Nerurkar Date: Thu, 3 Sep 2026 10:29:25 +0530 Subject: [PATCH 1/2] fix: stewardship re-upload dispersed replicas --- pkg/steward/steward.go | 17 +++++++++++++++ pkg/steward/steward_test.go | 43 +++++++++++++++++++++++++++++-------- 2 files changed, 51 insertions(+), 9 deletions(-) diff --git a/pkg/steward/steward.go b/pkg/steward/steward.go index 463d389814e..b1c66b94100 100644 --- a/pkg/steward/steward.go +++ b/pkg/steward/steward.go @@ -13,6 +13,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/file/redundancy" "github.com/ethersphere/bee/v2/pkg/postage" + "github.com/ethersphere/bee/v2/pkg/replicas" "github.com/ethersphere/bee/v2/pkg/retrieval" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/storer" @@ -77,6 +78,22 @@ func (s *steward) Reupload(ctx context.Context, root swarm.Address, stamper post ) } + if rLevel != redundancy.NONE { + rootChunk, err := getter.Get(ctx, root) + if err != nil { + return errors.Join(fmt.Errorf("get root chunk for dispersed replicas: %w", err), uploaderSession.Cleanup()) + } + + stamp, err := stamper.Stamp(rootChunk.Address(), rootChunk.Address()) + if err != nil { + return errors.Join(fmt.Errorf("stamping root chunk for dispersed replicas: %w", err), uploaderSession.Cleanup()) + } + + if err := replicas.NewPutter(uploaderSession, rLevel).Put(ctx, rootChunk.WithStamp(stamp)); err != nil { + return errors.Join(fmt.Errorf("re-uploading dispersed replicas: %w", err), uploaderSession.Cleanup()) + } + } + return uploaderSession.Done(root) } diff --git a/pkg/steward/steward_test.go b/pkg/steward/steward_test.go index 336729d9b12..758a0c7ff95 100644 --- a/pkg/steward/steward_test.go +++ b/pkg/steward/steward_test.go @@ -8,7 +8,6 @@ import ( "bytes" "context" "crypto/rand" - "errors" "sync" "sync/atomic" "testing" @@ -17,6 +16,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/file/pipeline/builder" "github.com/ethersphere/bee/v2/pkg/file/redundancy" postagetesting "github.com/ethersphere/bee/v2/pkg/postage/mock" + "github.com/ethersphere/bee/v2/pkg/soc" "github.com/ethersphere/bee/v2/pkg/steward" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/storage/inmemchunkstore" @@ -63,25 +63,36 @@ func TestSteward(t *testing.T) { } chunkCount := int(inmem.count.Load()) + replicaCount := redundancy.PARANOID.GetReplicaCount() + wantPushed := chunkCount + replicaCount done := make(chan struct{}) errc := make(chan error, 1) + replicaAddrs := make(map[string]struct{}) + var replicaMu sync.Mutex go func() { defer close(done) count := 0 for op := range store.PusherFeed() { - has, err := chunkStore.Has(ctx, op.Chunk.Address()) - if err != nil || !has { - if !has { - err = errors.New("chunk not found") - } + // DirectUpload only forwards pushed chunks over the feed; it does not + // persist them. Persist here so the post-reupload assertions (Has, + // IsRetrievable) observe pushed-but-not-yet-locally-known chunks the + // same way a real pushsync round-trip eventually would. + if err := chunkStore.Put(ctx, op.Chunk); err != nil { select { case errc <- err: default: } return } + + if sch, err := soc.FromChunk(op.Chunk); err == nil && bytes.Equal(sch.OwnerAddress(), swarm.ReplicasOwner) { + replicaMu.Lock() + replicaAddrs[op.Chunk.Address().String()] = struct{}{} + replicaMu.Unlock() + } + count++ - if count == chunkCount { + if count == wantPushed { return } } @@ -113,8 +124,22 @@ func TestSteward(t *testing.T) { } count := len(localRetrieval.retrievedChunks) - if count != chunkCount { - t.Fatalf("unexpected no of unique chunks retrieved: want %d have %d", chunkCount, count) + // IsRetrievable's root-chunk fetch goes through joiner -> replicas.NewGetter, which + // races the original root address against an initial batch of 2 replica candidate + // addresses before the first success cancels the rest (see replicas/getter.go). With + // real dispersed replicas now present (this is what this fix creates), up to 2 of + // those speculative replica fetches can also succeed and get recorded before + // cancellation lands, on top of the trie chunks retrieved by traversal. + const maxSpeculativeRootFetches = 2 + if count < chunkCount || count > chunkCount+maxSpeculativeRootFetches { + t.Fatalf("unexpected no of unique chunks retrieved: want between %d and %d, have %d", chunkCount, chunkCount+maxSpeculativeRootFetches, count) + } + + replicaMu.Lock() + gotReplicas := len(replicaAddrs) + replicaMu.Unlock() + if gotReplicas != replicaCount { + t.Fatalf("unexpected no of dispersed replicas re-uploaded: want %d have %d", replicaCount, gotReplicas) } } From 6a5de78fae1f80ffb3e7ba342cde2995f4688f46 Mon Sep 17 00:00:00 2001 From: Alok Nerurkar Date: Sat, 5 Sep 2026 17:14:28 +0530 Subject: [PATCH 2/2] fix: address review comments on dispersed replica reupload - Trim encrypted references to the 32-byte content address before deriving dispersed replica addresses; replicas are keyed on the plain content address, and root can be a 64-byte encrypted reference (address + decryption key). - Validate the fetched root chunk is a valid content-addressed chunk before deriving replicas from it. - Stamp each dispersed replica individually against its own SOC address as it is put, instead of stamping the root chunk once and reusing that stamp across all replicas (which have different addresses and would fail stamp validation on the receiving side). - Add TestStewardEncryptedReference covering both fixes, with a strict-address chunk store and a stamp-recording stamper to ensure both regressions are actually caught. --- pkg/steward/steward.go | 31 +++++-- pkg/steward/steward_test.go | 174 ++++++++++++++++++++++++++++++++++++ 2 files changed, 200 insertions(+), 5 deletions(-) diff --git a/pkg/steward/steward.go b/pkg/steward/steward.go index b1c66b94100..655c037e1d3 100644 --- a/pkg/steward/steward.go +++ b/pkg/steward/steward.go @@ -11,6 +11,8 @@ import ( "errors" "fmt" + "github.com/ethersphere/bee/v2/pkg/cac" + "github.com/ethersphere/bee/v2/pkg/encryption" "github.com/ethersphere/bee/v2/pkg/file/redundancy" "github.com/ethersphere/bee/v2/pkg/postage" "github.com/ethersphere/bee/v2/pkg/replicas" @@ -79,17 +81,36 @@ func (s *steward) Reupload(ctx context.Context, root swarm.Address, stamper post } if rLevel != redundancy.NONE { - rootChunk, err := getter.Get(ctx, root) + // Dispersed replicas are keyed on the 32-byte content address. root can be + // an encrypted reference (address + decryption key), so trim it before + // deriving replica addresses, or they won't match what a downloader + // deriving replicas from the plain address expects. + contentAddr := root + if len(root.Bytes()) == encryption.ReferenceSize { + contentAddr = swarm.NewAddress(root.Bytes()[:swarm.HashSize]) + } + + rootChunk, err := getter.Get(ctx, contentAddr) if err != nil { return errors.Join(fmt.Errorf("get root chunk for dispersed replicas: %w", err), uploaderSession.Cleanup()) } - stamp, err := stamper.Stamp(rootChunk.Address(), rootChunk.Address()) - if err != nil { - return errors.Join(fmt.Errorf("stamping root chunk for dispersed replicas: %w", err), uploaderSession.Cleanup()) + if !cac.Valid(rootChunk) { + return errors.Join(fmt.Errorf("root chunk %s is not a valid content-addressed chunk", contentAddr), uploaderSession.Cleanup()) } - if err := replicas.NewPutter(uploaderSession, rLevel).Put(ctx, rootChunk.WithStamp(stamp)); err != nil { + // Stamp each replica individually as it is put, keyed on its own SOC + // address - not the root chunk's address, which replicas.NewPutter + // wraps into a differently-addressed SOC chunk per replica. + stampedPutter := storage.PutterFunc(func(ctx context.Context, ch swarm.Chunk) error { + stamp, err := stamper.Stamp(ch.Address(), ch.Address()) + if err != nil { + return fmt.Errorf("stamping replica %s: %w", ch.Address(), err) + } + return uploaderSession.Put(ctx, ch.WithStamp(stamp)) + }) + + if err := replicas.NewPutter(stampedPutter, rLevel).Put(ctx, rootChunk); err != nil { return errors.Join(fmt.Errorf("re-uploading dispersed replicas: %w", err), uploaderSession.Cleanup()) } } diff --git a/pkg/steward/steward_test.go b/pkg/steward/steward_test.go index 758a0c7ff95..25330df06b4 100644 --- a/pkg/steward/steward_test.go +++ b/pkg/steward/steward_test.go @@ -8,6 +8,7 @@ import ( "bytes" "context" "crypto/rand" + "fmt" "sync" "sync/atomic" "testing" @@ -15,6 +16,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/file/pipeline/builder" "github.com/ethersphere/bee/v2/pkg/file/redundancy" + "github.com/ethersphere/bee/v2/pkg/postage" postagetesting "github.com/ethersphere/bee/v2/pkg/postage/mock" "github.com/ethersphere/bee/v2/pkg/soc" "github.com/ethersphere/bee/v2/pkg/steward" @@ -34,6 +36,33 @@ func (c *counter) Put(ctx context.Context, ch swarm.Chunk) (err error) { return c.ChunkStore.Put(ctx, ch) } +// recordingStamper wraps a postage.Stamper and records the address each Stamp +// call was made for, so tests can assert every uploaded chunk (including each +// dispersed replica) was stamped against its own address rather than a single +// shared stamp computed once for the root chunk. +type recordingStamper struct { + postage.Stamper + mu sync.Mutex + stamped map[string]int +} + +func newRecordingStamper() *recordingStamper { + return &recordingStamper{Stamper: postagetesting.NewStamper(), stamped: make(map[string]int)} +} + +func (r *recordingStamper) Stamp(addr, idAddr swarm.Address) (*postage.Stamp, error) { + r.mu.Lock() + r.stamped[addr.String()]++ + r.mu.Unlock() + return r.Stamper.Stamp(addr, idAddr) +} + +func (r *recordingStamper) stampedFor(addr swarm.Address) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.stamped[addr.String()] +} + func TestSteward(t *testing.T) { t.Parallel() inmem := &counter{ChunkStore: inmemchunkstore.New()} @@ -143,6 +172,151 @@ func TestSteward(t *testing.T) { } } +// strictAddressChunkStore wraps a storage.ChunkStore and requires Get to be +// called with an exact 32-byte content address - unlike inmemchunkstore, which +// silently truncates longer (e.g. 64-byte encrypted) addresses to the first 32 +// bytes on lookup, masking a caller that forgets to trim an encrypted reference +// before deriving replica addresses from it. +type strictAddressChunkStore struct { + storage.ChunkStore +} + +func (s *strictAddressChunkStore) Get(ctx context.Context, addr swarm.Address) (swarm.Chunk, error) { + if len(addr.Bytes()) != swarm.HashSize { + return nil, fmt.Errorf("strictAddressChunkStore: Get called with non-content address %s (len %d)", addr, len(addr.Bytes())) + } + return s.ChunkStore.Get(ctx, addr) +} + +// TestStewardEncryptedReference verifies that Reupload correctly derives dispersed +// replica addresses from an encrypted reference (address + decryption key), by +// trimming it to the 32-byte content address before deriving replicas - otherwise +// the replica addresses computed would not match what a downloader deriving +// replicas from the plain content address expects. +func TestStewardEncryptedReference(t *testing.T) { + t.Parallel() + inmem := &counter{ChunkStore: &strictAddressChunkStore{ChunkStore: inmemchunkstore.New()}} + + var ( + ctx = context.Background() + chunks = 3 + data = make([]byte, chunks*4096) + chunkStore = inmem + store = mockstorer.NewWithChunkStore(chunkStore) + s = steward.New(store, &localRetriever{ChunkStore: chunkStore}, inmem) + stamper = newRecordingStamper() + ) + n, err := rand.Read(data) + if n != cap(data) { + t.Fatal("short read") + } + if err != nil { + t.Fatal(err) + } + + pipe := builder.NewPipelineBuilder(ctx, chunkStore, true, redundancy.NONE) + addr, err := builder.FeedPipeline(ctx, pipe, bytes.NewReader(data)) + if err != nil { + t.Fatal(err) + } + if len(addr.Bytes()) != swarm.HashSize+32 { + t.Fatalf("expected an encrypted reference of length %d, got %d", swarm.HashSize+32, len(addr.Bytes())) + } + + replicaCount := redundancy.PARANOID.GetReplicaCount() + contentAddr := swarm.NewAddress(addr.Bytes()[:swarm.HashSize]) + + replicaAddrs := make(map[string]struct{}) + var replicaMu sync.Mutex + done := make(chan struct{}) + errc := make(chan error, 1) + wantPushed := int(inmem.count.Load()) + replicaCount + go func() { + defer close(done) + count := 0 + for op := range store.PusherFeed() { + if err := chunkStore.Put(ctx, op.Chunk); err != nil { + select { + case errc <- err: + default: + } + return + } + if sch, err := soc.FromChunk(op.Chunk); err == nil && bytes.Equal(sch.OwnerAddress(), swarm.ReplicasOwner) { + replicaMu.Lock() + replicaAddrs[op.Chunk.Address().String()] = struct{}{} + replicaMu.Unlock() + } + count++ + if count == wantPushed { + return + } + } + }() + + err = s.Reupload(ctx, addr, stamper, redundancy.PARANOID) + if err != nil { + t.Fatal(err) + } + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("took too long to finish") + } + select { + case err := <-errc: + t.Fatalf("unexpected error: %v", err) + default: + } + + replicaMu.Lock() + gotReplicas := len(replicaAddrs) + replicaMu.Unlock() + if gotReplicas != replicaCount { + t.Fatalf("unexpected no of dispersed replicas re-uploaded: want %d have %d", replicaCount, gotReplicas) + } + + // Every replica must wrap the plain 32-byte content address's chunk, and + // replicas.NewPutter derives replica addresses from that same chunk's + // address (ch.Address()) - so this also proves replica addresses were + // derived from contentAddr, not the 64-byte encrypted reference. If the + // reference had not been trimmed before the fix, this lookup would have + // failed (get root chunk for dispersed replicas) or wrapped the wrong chunk. + for addrStr := range replicaAddrs { + replicaAddr := swarm.MustParseHexAddress(addrStr) + sch, err := chunkStore.Get(ctx, replicaAddr) + if err != nil { + t.Fatalf("get replica chunk %s: %v", replicaAddr, err) + } + replicaSOC, err := soc.FromChunk(sch) + if err != nil { + t.Fatalf("replica %s is not a valid SOC chunk: %v", replicaAddr, err) + } + if !replicaSOC.WrappedChunk().Address().Equal(contentAddr) { + t.Fatalf("replica %s wraps chunk %s, want %s", replicaAddr, replicaSOC.WrappedChunk().Address(), contentAddr) + } + + // Each replica must be individually stamped against its own SOC + // address - not stamped once against the root chunk's address and + // reused, which would fail stamp validation on the receiving side + // since a postage stamp is only valid for the specific address it + // was computed against. + if got := stamper.stampedFor(replicaAddr); got != 1 { + t.Fatalf("replica %s: want exactly 1 Stamp call for its own address, got %d", replicaAddr, got) + } + } + // The root chunk's own address gets stamped exactly once via the normal + // traversal path (fn), because it's re-uploaded as part of the trie like any + // other chunk. It must not be stamped a second time by the replica-upload + // step: reusing that stamp on a differently-addressed SOC replica chunk + // would fail stamp validation on the receiving side, since a stamp is only + // valid for the specific address it was computed against. + if got := stamper.stampedFor(contentAddr); got != 1 { + t.Fatalf("root chunk address %s: want exactly 1 Stamp call (from trie traversal), got %d", contentAddr, got) + } +} + type localRetriever struct { storage.ChunkStore mu sync.Mutex