swarm: Chunk refactor (#17659)

Co-authored-by: Janos Guljas <janos@resenje.org>
Co-authored-by: Balint Gabor <balint.g@gmail.com>
Co-authored-by: Anton Evangelatov <anton.evangelatov@gmail.com>
Co-authored-by: Viktor Trón <viktor.tron@gmail.com>

1 files changed, 19 insertions, 31 deletions

diff --git a/swarm/swarm.go b/swarm/swarm.go
index baf71b962..13aa1125d 100644
--- a/swarm/swarm.go
+++ b/swarm/swarm.go
@@ -75,8 +75,8 @@ type Swarm struct {
 	privateKey  *ecdsa.PrivateKey
 	corsString  string
 	swapEnabled bool
-	lstore      *storage.LocalStore // local store, needs to store for releasing resources after node stopped
-	sfs         *fuse.SwarmFS       // need this to cleanup all the active mounts on node exit
+	netStore    *storage.NetStore
+	sfs         *fuse.SwarmFS // need this to cleanup all the active mounts on node exit
 	ps          *pss.Pss
 
 	tracerClose io.Closer
@@ -164,37 +164,40 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 		self.dns = resolver
 	}
 
-	self.lstore, err = storage.NewLocalStore(config.LocalStoreParams, mockStore)
+	lstore, err := storage.NewLocalStore(config.LocalStoreParams, mockStore)
 	if err != nil {
-		return
+		return nil, err
+	}
+
+	self.netStore, err = storage.NewNetStore(lstore, nil)
+	if err != nil {
+		return nil, err
 	}
 
-	db := storage.NewDBAPI(self.lstore)
 	to := network.NewKademlia(
 		common.FromHex(config.BzzKey),
 		network.NewKadParams(),
 	)
-	delivery := stream.NewDelivery(to, db)
+	delivery := stream.NewDelivery(to, self.netStore)
+	self.netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, config.DeliverySkipCheck).New
 
-	self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, &stream.RegistryOptions{
-		SkipCheck:       config.DeliverySkipCheck,
+	self.streamer = stream.NewRegistry(addr, delivery, self.netStore, stateStore, &stream.RegistryOptions{
+		SkipCheck:       config.SyncingSkipCheck,
 		DoSync:          config.SyncEnabled,
 		DoRetrieve:      true,
 		SyncUpdateDelay: config.SyncUpdateDelay,
 	})
 
-	// set up NetStore, the cloud storage local access layer
-	netStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve)
 	// Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage
-	self.fileStore = storage.NewFileStore(netStore, self.config.FileStoreParams)
+	self.fileStore = storage.NewFileStore(self.netStore, self.config.FileStoreParams)
 
 	var resourceHandler *mru.Handler
 	rhparams := &mru.HandlerParams{}
 
 	resourceHandler = mru.NewHandler(rhparams)
-	resourceHandler.SetStore(netStore)
+	resourceHandler.SetStore(self.netStore)
 
-	self.lstore.Validators = []storage.ChunkValidator{
+	lstore.Validators = []storage.ChunkValidator{
 		storage.NewContentAddressValidator(storage.MakeHashFunc(storage.DefaultHash)),
 		resourceHandler,
 	}
@@ -399,7 +402,7 @@ func (self *Swarm) periodicallyUpdateGauges() {
 
 func (self *Swarm) updateGauges() {
 	uptimeGauge.Update(time.Since(startTime).Nanoseconds())
-	requestsCacheGauge.Update(int64(self.lstore.RequestsCacheLen()))
+	requestsCacheGauge.Update(int64(self.netStore.RequestsCacheLen()))
 }
 
 // implements the node.Service interface
@@ -420,8 +423,8 @@ func (self *Swarm) Stop() error {
 		ch.Save()
 	}
 
-	if self.lstore != nil {
-		self.lstore.DbStore.Close()
+	if self.netStore != nil {
+		self.netStore.Close()
 	}
 	self.sfs.Stop()
 	stopCounter.Inc(1)
@@ -478,21 +481,6 @@ func (self *Swarm) APIs() []rpc.API {
 			Service:   self.sfs,
 			Public:    false,
 		},
-		// storage APIs
-		// DEPRECATED: Use the HTTP API instead
-		{
-			Namespace: "bzz",
-			Version:   "0.1",
-			Service:   api.NewStorage(self.api),
-			Public:    true,
-		},
-		{
-			Namespace: "bzz",
-			Version:   "0.1",
-			Service:   api.NewFileSystem(self.api),
-			Public:    false,
-		},
-		// {Namespace, Version, api.NewAdmin(self), false},
 	}
 
 	apis = append(apis, self.bzz.APIs()...)