feat: soc dispersed replica v2

.github/workflows/beekeeper.yml

@@ -14,7 +14,7 @@ env:
   SETUP_CONTRACT_IMAGE: "ethersphere/bee-localchain"
   SETUP_CONTRACT_IMAGE_TAG: "0.9.4"
   BEELOCAL_BRANCH: "main"
-  BEEKEEPER_BRANCH: "master"
+  BEEKEEPER_BRANCH: "feat/soc-dispersed"
   BEEKEEPER_METRICS_ENABLED: false
   REACHABILITY_OVERRIDE_PUBLIC: true
   BATCHFACTOR_OVERRIDE_PUBLIC: 2

openapi/Swarm.yaml

@@ -879,6 +879,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.
@@ -926,6 +927,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"
@@ -977,6 +979,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
@@ -1035,6 +1038,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/accesscontrol.go

@@ -126,7 +126,7 @@ func (s *Service) actDecryptionHandler() func(h http.Handler) http.Handler {
 				cache = *headers.Cache
 			}
 			ctx := r.Context()
-			ls := loadsave.NewReadonly(s.storer.Download(cache), s.storer.Cache(), redundancy.DefaultLevel)
+			ls := loadsave.NewReadonly(s.storer.Download(cache), s.storer.Cache(), redundancy.PARANOID)
 			reference, err := s.accesscontrol.DownloadHandler(ctx, ls, paths.Address, headers.Publisher, *headers.HistoryAddress, timestamp)
 			if err != nil {
 				logger.Debug("access control download failed", "error", err)
@@ -159,7 +159,7 @@ func (s *Service) actEncryptionHandler(
 	historyRootHash swarm.Address,
 ) (swarm.Address, swarm.Address, error) {
 	publisherPublicKey := &s.publicKey
-	ls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, false, redundancy.NONE), redundancy.DefaultLevel)
+	ls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, false, redundancy.NONE), redundancy.PARANOID)
 	storageReference, historyReference, encryptedReference, err := s.accesscontrol.UploadHandler(ctx, ls, reference, publisherPublicKey, historyRootHash)
 	if err != nil {
 		return swarm.ZeroAddress, swarm.ZeroAddress, err
@@ -204,7 +204,7 @@ func (s *Service) actListGranteesHandler(w http.ResponseWriter, r *http.Request)
 		cache = *headers.Cache
 	}
 	publisher := &s.publicKey
-	ls := loadsave.NewReadonly(s.storer.Download(cache), s.storer.Cache(), redundancy.DefaultLevel)
+	ls := loadsave.NewReadonly(s.storer.Download(cache), s.storer.Cache(), redundancy.PARANOID)
 	grantees, err := s.accesscontrol.Get(r.Context(), ls, publisher, paths.GranteesAddress)
 	if err != nil {
 		logger.Debug("could not get grantees", "error", err)
@@ -344,8 +344,8 @@ func (s *Service) actGrantRevokeHandler(w http.ResponseWriter, r *http.Request)
 
 	granteeref := paths.GranteesAddress
 	publisher := &s.publicKey
-	ls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, false, redundancy.NONE), redundancy.DefaultLevel)
-	gls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, granteeListEncrypt, redundancy.NONE), redundancy.DefaultLevel)
+	ls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, false, redundancy.NONE), redundancy.PARANOID)
+	gls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, granteeListEncrypt, redundancy.NONE), redundancy.PARANOID)
 	granteeref, encryptedglref, historyref, actref, err := s.accesscontrol.UpdateHandler(ctx, ls, gls, granteeref, historyAddress, publisher, grantees.Addlist, grantees.Revokelist)
 	if err != nil {
 		logger.Debug("failed to update grantee list", "error", err)
@@ -498,8 +498,8 @@ func (s *Service) actCreateGranteesHandler(w http.ResponseWriter, r *http.Reques
 	}
 
 	publisher := &s.publicKey
-	ls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, false, redundancy.NONE), redundancy.DefaultLevel)
-	gls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, granteeListEncrypt, redundancy.NONE), redundancy.DefaultLevel)
+	ls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, false, redundancy.NONE), redundancy.PARANOID)
+	gls := loadsave.New(s.storer.Download(true), s.storer.Cache(), requestPipelineFactory(ctx, putter, granteeListEncrypt, redundancy.NONE), redundancy.PARANOID)
 	granteeref, encryptedglref, historyref, actref, err := s.accesscontrol.UpdateHandler(ctx, ls, gls, swarm.ZeroAddress, historyAddress, publisher, list, nil)
 	if err != nil {
 		logger.Debug("failed to create grantee list", "error", err)

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"
@@ -60,6 +61,13 @@ func lookaheadBufferSize(size int64) int {
 	return largeFileBufferSize
 }
 
+func getRedundancyLevel(rLevel *redundancy.Level) redundancy.Level {
+	if rLevel != nil {
+		return *rLevel
+	}
+	return redundancy.PARANOID
+}
+
 func (s *Service) bzzUploadHandler(w http.ResponseWriter, r *http.Request) {
 	span, logger, ctx := s.tracer.StartSpanFromContext(r.Context(), "post_bzz", s.logger.WithName("post_bzz").Build())
 	defer span.Finish()
@@ -400,13 +408,11 @@ func (s *Service) serveReference(logger log.Logger, address swarm.Address, pathV
 		cache = *headers.Cache
 	}
 
-	rLevel := redundancy.DefaultLevel
-	if headers.RLevel != nil {
-		rLevel = *headers.RLevel
-	}
+	rLevel := getRedundancyLevel(headers.RLevel)
 
 	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 +440,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 {
@@ -622,10 +628,7 @@ func (s *Service) downloadHandler(logger log.Logger, w http.ResponseWriter, r *h
 		jsonhttp.BadRequest(w, "could not parse headers")
 		return
 	}
-	rLevel := redundancy.DefaultLevel
-	if headers.RLevel != nil {
-		rLevel = *headers.RLevel
-	}
+	rLevel := getRedundancyLevel(headers.RLevel)
 
 	var (
 		reader file.Joiner
@@ -701,6 +704,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 +737,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, feeds.WithGetter(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"
@@ -806,6 +811,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/chunk.go

@@ -260,8 +260,8 @@ func (s *Service) chunkGetHandler(w http.ResponseWriter, r *http.Request) {
 			loggerV1.Debug("chunk not found", "address", address)
 			jsonhttp.NotFound(w, "chunk not found")
 			return
-
 		}
+
 		logger.Debug("read chunk failed", "chunk_address", address, "error", err)
 		logger.Error(nil, "read chunk failed")
 		jsonhttp.InternalServerError(w, "read chunk failed")