feat: soc dispersed replica

openapi/Swarm.yaml

@@ -885,6 +885,7 @@ paths:
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPostageStamp"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmAct"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmActHistoryAddress"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyLevelParameter"
       requestBody:
         required: true
         description: The SOC binary data is composed of the span (8 bytes) and the at most 4KB payload.
@@ -932,6 +933,7 @@ paths:
           description: Arbitrary identifier of the related data
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmOnlyRootChunkParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCache"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyLevelParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyStrategyParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyFallbackModeParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmChunkRetrievalTimeoutParameter"
@@ -983,6 +985,7 @@ paths:
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPostageBatchId"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmAct"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmActHistoryAddress"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyLevelParameter"
       responses:
         "201":
           description: Created
@@ -1041,6 +1044,7 @@ paths:
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmFeedLegacyResolve"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmOnlyRootChunkParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCache"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyLevelParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyStrategyParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyFallbackModeParameter"
         - $ref: "SwarmCommon.yaml#/components/parameters/SwarmChunkRetrievalTimeoutParameter"

pkg/api/bzz.go

@@ -32,6 +32,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/manifest"
 	"github.com/ethersphere/bee/v2/pkg/postage"
+	"github.com/ethersphere/bee/v2/pkg/replicas"
 	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/storer"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -406,7 +407,8 @@ func (s *Service) serveReference(logger log.Logger, address swarm.Address, pathV
 	}
 
 	ctx := r.Context()
-	ls := loadsave.NewReadonly(s.storer.Download(cache), s.storer.Cache(), redundancy.DefaultLevel)
+	g := s.storer.Download(cache)
+	ls := loadsave.NewReadonly(g, s.storer.Cache(), rLevel)
 	feedDereferenced := false
 
 	ctx, err := getter.SetConfigInContext(ctx, headers.Strategy, headers.FallbackMode, headers.ChunkRetrievalTimeout, logger)
@@ -434,7 +436,7 @@ FETCH:
 	// unmarshal as mantaray first and possibly resolve the feed, otherwise
 	// go on normally.
 	if !feedDereferenced {
-		if l, err := s.manifestFeed(ctx, m); err == nil {
+		if l, err := s.manifestFeed(ctx, m, replicas.NewSocGetter(g, rLevel)); err == nil {
 			// we have a feed manifest here
 			ch, cur, _, err := l.At(ctx, time.Now().Unix(), 0)
 			if err != nil {
@@ -701,6 +703,7 @@ func manifestMetadataLoad(
 func (s *Service) manifestFeed(
 	ctx context.Context,
 	m manifest.Interface,
+	st storage.Getter,
 ) (feeds.Lookup, error) {
 	e, err := m.Lookup(ctx, "/")
 	if err != nil {
@@ -733,5 +736,5 @@ func (s *Service) manifestFeed(
 		return nil, fmt.Errorf("node lookup: %s", "feed metadata absent")
 	}
 	f := feeds.New(topic, common.BytesToAddress(owner))
-	return s.feedFactory.NewLookup(*t, f)
+	return s.feedFactory.NewLookup(*t, f, st)
 }

pkg/api/bzz_test.go

@@ -7,6 +7,7 @@ package api_test
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
@@ -19,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/ethersphere/bee/v2/pkg/api"
+	"github.com/ethersphere/bee/v2/pkg/feeds"
 	"github.com/ethersphere/bee/v2/pkg/file/loadsave"
 	"github.com/ethersphere/bee/v2/pkg/file/redundancy"
 	"github.com/ethersphere/bee/v2/pkg/jsonhttp"
@@ -27,7 +29,10 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/manifest"
 	mockbatchstore "github.com/ethersphere/bee/v2/pkg/postage/batchstore/mock"
 	mockpost "github.com/ethersphere/bee/v2/pkg/postage/mock"
+	"github.com/ethersphere/bee/v2/pkg/replicas"
+	"github.com/ethersphere/bee/v2/pkg/soc"
 	testingsoc "github.com/ethersphere/bee/v2/pkg/soc/testing"
+	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/storage/inmemchunkstore"
 	mockstorer "github.com/ethersphere/bee/v2/pkg/storer/mock"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -809,6 +814,161 @@ func TestFeedIndirection(t *testing.T) {
 		t.Fatalf("expected file reference, did not got any")
 	}
 
+	// get root chunk of data
+	// and wrap it in a feed
+	rootCh, err := storer.ChunkStore().Get(context.Background(), resp.Reference)
+	if err != nil {
+		t.Fatal(err)
+	}
+	socRootCh := testingsoc.GenerateMockSOC(t, rootCh.Data()[swarm.SpanSize:]).Chunk()
+
+	// now use the "content" to mock the feed lookup
+	// also, use the mocked mantaray chunks that unmarshal
+	// into a real manifest with the mocked feed values when
+	// called from the bzz endpoint. then call the bzz endpoint with
+	// the pregenerated feed root manifest hash
+
+	t.Run("feed wrapping", func(t *testing.T) {
+		var (
+			look                = newMockLookup(-1, 0, socRootCh, nil, &id{}, nil)
+			factory             = newMockFactory(look)
+			bzzDownloadResource = func(addr, path string) string { return "/bzz/" + addr + "/" + path }
+			ctx                 = context.Background()
+		)
+		client, _, _, _ = newTestServer(t, testServerOptions{
+			Storer: storer,
+			Logger: logger,
+			Feeds:  factory,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		m, err := manifest.NewDefaultManifest(
+			loadsave.New(storer.ChunkStore(), storer.Cache(), pipelineFactory(storer.Cache(), false, 0), redundancy.DefaultLevel),
+			false,
+		)
+		if err != nil {
+			t.Fatal(err)
+		}
+		emptyAddr := make([]byte, 32)
+		err = m.Add(ctx, manifest.RootPath, manifest.NewEntry(swarm.NewAddress(emptyAddr), map[string]string{
+			api.FeedMetadataEntryOwner: "8d3766440f0d7b949a5e32995d09619a7f86e632",
+			api.FeedMetadataEntryTopic: "abcc",
+			api.FeedMetadataEntryType:  "epoch",
+		}))
+		if err != nil {
+			t.Fatal(err)
+		}
+		manifRef, err := m.Store(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		jsonhttptest.Request(t, client, http.MethodGet, bzzDownloadResource(manifRef.String(), ""), http.StatusOK,
+			jsonhttptest.WithExpectedResponse(updateData),
+			jsonhttptest.WithExpectedContentLength(len(updateData)),
+			jsonhttptest.WithExpectedResponseHeader(api.AccessControlExposeHeaders, api.SwarmFeedIndexHeader),
+			jsonhttptest.WithExpectedResponseHeader(api.AccessControlExposeHeaders, api.ContentDispositionHeader),
+			jsonhttptest.WithExpectedResponseHeader(api.ContentDispositionHeader, `inline; filename="index.html"`),
+			jsonhttptest.WithExpectedResponseHeader(api.ContentTypeHeader, "text/html; charset=utf-8"),
+		)
+	})
+
+	t.Run("redundancy", func(t *testing.T) {
+		// enough to test two redundancy levels since
+		tests := []struct {
+			name   string
+			rLevel redundancy.Level
+		}{
+			{
+				name:   "none",
+				rLevel: redundancy.NONE,
+			},
+			{
+				name:   "medium",
+				rLevel: redundancy.MEDIUM,
+			},
+		}
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				rLevel := tt.rLevel
+				socRoot, _ := soc.FromChunk(socRootCh)
+				socPutter := replicas.NewSocPutter(storer, rLevel)
+				err = socPutter.Put(context.Background(), socRootCh)
+				if err != nil {
+					t.Fatalf("failed to put SOC chunk with redundancy: %v", err)
+				}
+
+				m, err := manifest.NewDefaultManifest(
+					loadsave.New(storer.ChunkStore(), storer.Cache(), pipelineFactory(storer.Cache(), false, rLevel), rLevel),
+					false,
+				)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				// Add the feed entry to the manifest
+				hexId := hex.EncodeToString(socRoot.ID())
+				hexOwner := hex.EncodeToString(socRoot.OwnerAddress())
+				err = m.Add(context.Background(), manifest.RootPath, manifest.NewEntry(socRootCh.Address(), map[string]string{
+					api.FeedMetadataEntryOwner: hexOwner,
+					api.FeedMetadataEntryTopic: hexId,
+					api.FeedMetadataEntryType:  "sequence",
+				}))
+				if err != nil {
+					t.Fatal(err)
+				}
+				manifestRef, err := m.Store(context.Background())
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				// Create mockLookup and mockFactory for feed
+				look := newRedundancyMockLookup(
+					rLevel,
+					storer.ChunkStore(),
+					func() (swarm.Chunk, feeds.Index, feeds.Index) {
+						return socRootCh, &id{}, &id{}
+					},
+				)
+				feedFactory := newMockFactory(look)
+
+				// Update the test server with the feed factory
+				client, _, _, _ := newTestServer(t, testServerOptions{
+					Storer: storer,
+					Logger: log.Noop,
+					Post:   mockpost.New(mockpost.WithAcceptAll()),
+					Feeds:  feedFactory,
+				})
+
+				// remove original chunk from store
+				cs, ok := storer.ChunkStore().(storage.ChunkStore)
+				if !ok {
+					t.Fatalf("chunk store not available for deletion")
+				}
+				err = cs.Delete(context.Background(), socRootCh.Address())
+				if err != nil {
+					t.Fatalf("Failed to delete soc chunk: %v", err)
+				}
+
+				manifestHex := manifestRef.String()
+
+				if rLevel == redundancy.NONE {
+					jsonhttptest.Request(t, client, http.MethodGet, "/bzz/"+manifestHex+"/", http.StatusNotFound)
+					return
+				}
+				jsonhttptest.Request(t, client, http.MethodGet, "/bzz/"+manifestHex+"/", http.StatusOK,
+					jsonhttptest.WithExpectedResponse(updateData),
+					jsonhttptest.WithExpectedContentLength(len(updateData)),
+					jsonhttptest.WithExpectedResponseHeader(api.AccessControlExposeHeaders, api.SwarmFeedIndexHeader),
+					jsonhttptest.WithExpectedResponseHeader(api.AccessControlExposeHeaders, api.ContentDispositionHeader),
+					jsonhttptest.WithExpectedResponseHeader(api.ContentTypeHeader, "text/html; charset=utf-8"),
+					jsonhttptest.WithExpectedResponseHeader(api.ContentDispositionHeader, `inline; filename="index.html"`),
+				)
+			})
+		}
+	})
+
 	m, err := manifest.NewDefaultManifest(
 		loadsave.New(storer.ChunkStore(), storer.Cache(), pipelineFactory(storer.Cache(), false, 0), redundancy.DefaultLevel),
 		false,

pkg/api/feed.go

@@ -23,6 +23,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/manifest/mantaray"
 	"github.com/ethersphere/bee/v2/pkg/manifest/simple"
 	"github.com/ethersphere/bee/v2/pkg/postage"
+	"github.com/ethersphere/bee/v2/pkg/replicas"
 	"github.com/ethersphere/bee/v2/pkg/soc"
 	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/storer"
@@ -66,15 +67,21 @@ func (s *Service) feedGetHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	headers := struct {
-		OnlyRootChunk bool `map:"Swarm-Only-Root-Chunk"`
+		OnlyRootChunk   bool             `map:"Swarm-Only-Root-Chunk"`
+		RedundancyLevel redundancy.Level `map:"Swarm-Redundancy-Level"`
 	}{}
 	if response := s.mapStructure(r.Header, &headers); response != nil {
 		response("invalid header params", logger, w)
 		return
 	}
 
+	getter := s.storer.Download(false)
+	if headers.RedundancyLevel > redundancy.NONE {
+		getter = replicas.NewSocGetter(getter, headers.RedundancyLevel)
+	}
+
 	f := feeds.New(paths.Topic, paths.Owner)
-	lookup, err := s.feedFactory.NewLookup(feeds.Sequence, f)
+	lookup, err := s.feedFactory.NewLookup(feeds.Sequence, f, getter)
 	if err != nil {
 		logger.Debug("new lookup failed", "owner", paths.Owner, "error", err)
 		logger.Error(nil, "new lookup failed")
@@ -170,11 +177,12 @@ func (s *Service) feedPostHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	headers := struct {
-		BatchID        []byte        `map:"Swarm-Postage-Batch-Id" validate:"required"`
-		Pin            bool          `map:"Swarm-Pin"`
-		Deferred       *bool         `map:"Swarm-Deferred-Upload"`
-		Act            bool          `map:"Swarm-Act"`
-		HistoryAddress swarm.Address `map:"Swarm-Act-History-Address"`
+		BatchID         []byte           `map:"Swarm-Postage-Batch-Id" validate:"required"`
+		Pin             bool             `map:"Swarm-Pin"`
+		Deferred        *bool            `map:"Swarm-Deferred-Upload"`
+		Act             bool             `map:"Swarm-Act"`
+		HistoryAddress  swarm.Address    `map:"Swarm-Act-History-Address"`
+		RedundancyLevel redundancy.Level `map:"Swarm-Redundancy-Level"`
 	}{}
 	if response := s.mapStructure(r.Header, &headers); response != nil {
 		response("invalid header params", logger, w)
@@ -231,7 +239,9 @@ func (s *Service) feedPostHandler(w http.ResponseWriter, r *http.Request) {
 		logger:         logger,
 	}
 
-	l := loadsave.New(s.storer.ChunkStore(), s.storer.Cache(), requestPipelineFactory(r.Context(), putter, false, 0), redundancy.DefaultLevel)
+	rLevel := headers.RedundancyLevel
+
+	l := loadsave.New(s.storer.ChunkStore(), s.storer.Cache(), requestPipelineFactory(r.Context(), putter, false, 0), rLevel)
 	feedManifest, err := manifest.NewDefaultManifest(l, false)
 	if err != nil {
 		logger.Debug("create manifest failed", "error", err)