diff --git a/pkg/steward/steward.go b/pkg/steward/steward.go index 463d389814e..655c037e1d3 100644 --- a/pkg/steward/steward.go +++ b/pkg/steward/steward.go @@ -11,8 +11,11 @@ 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" "github.com/ethersphere/bee/v2/pkg/retrieval" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/storer" @@ -77,6 +80,41 @@ func (s *steward) Reupload(ctx context.Context, root swarm.Address, stamper post ) } + if rLevel != redundancy.NONE { + // 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()) + } + + if !cac.Valid(rootChunk) { + return errors.Join(fmt.Errorf("root chunk %s is not a valid content-addressed chunk", contentAddr), uploaderSession.Cleanup()) + } + + // 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()) + } + } + return uploaderSession.Done(root) } diff --git a/pkg/steward/steward_test.go b/pkg/steward/steward_test.go index 336729d9b12..25330df06b4 100644 --- a/pkg/steward/steward_test.go +++ b/pkg/steward/steward_test.go @@ -8,7 +8,7 @@ import ( "bytes" "context" "crypto/rand" - "errors" + "fmt" "sync" "sync/atomic" "testing" @@ -16,7 +16,9 @@ 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" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/storage/inmemchunkstore" @@ -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()} @@ -63,25 +92,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 +153,167 @@ 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) + } +} + +// 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) } }