Integrate encryption into dpa and chunker

cmd/swarm/config_test.go

@@ -157,7 +157,6 @@ func TestConfigFileOverrides(t *testing.T) {
 	defaultConf.NetworkId = 54
 	defaultConf.Port = httpPort
 	defaultConf.StoreParams.DbCapacity = 9000000
-	defaultConf.ChunkerParams.Branches = 64
 	defaultConf.HiveParams.KeepAliveInterval = 6000000000
 	defaultConf.Swap.Params.Strategy.AutoCashInterval = 600 * time.Second
 	//defaultConf.SyncParams.KeyBufferSize = 512
@@ -237,10 +236,6 @@ func TestConfigFileOverrides(t *testing.T) {
 		t.Fatalf("Expected network ID to be %d, got %d", 54, info.NetworkId)
 	}
 
-	if info.ChunkerParams.Branches != 64 {
-		t.Fatalf("Expected chunker params branches to be %d, got %d", 64, info.ChunkerParams.Branches)
-	}
-
 	if info.HiveParams.KeepAliveInterval != 6000000000 {
 		t.Fatalf("Expected HiveParams KeepAliveInterval to be %d, got %d", uint64(6000000000), uint64(info.HiveParams.KeepAliveInterval))
 	}
@@ -374,7 +369,6 @@ func TestConfigCmdLineOverridesFile(t *testing.T) {
 	defaultConf.NetworkId = 54
 	defaultConf.Port = "8588"
 	defaultConf.StoreParams.DbCapacity = 9000000
-	defaultConf.ChunkerParams.Branches = 64
 	defaultConf.HiveParams.KeepAliveInterval = 6000000000
 	defaultConf.Swap.Params.Strategy.AutoCashInterval = 600 * time.Second
 	//defaultConf.SyncParams.KeyBufferSize = 512
@@ -456,10 +450,6 @@ func TestConfigCmdLineOverridesFile(t *testing.T) {
 		t.Fatalf("Expected network ID to be %d, got %d", 54, info.NetworkId)
 	}
 
-	if info.ChunkerParams.Branches != 64 {
-		t.Fatalf("Expected chunker params branches to be %d, got %d", 64, info.ChunkerParams.Branches)
-	}
-
 	if info.HiveParams.KeepAliveInterval != 6000000000 {
 		t.Fatalf("Expected HiveParams KeepAliveInterval to be %d, got %d", uint64(6000000000), uint64(info.HiveParams.KeepAliveInterval))
 	}

cmd/swarm/hash.go

@@ -38,8 +38,8 @@ func hash(ctx *cli.Context) {
 	defer f.Close()
 
 	stat, _ := f.Stat()
-	chunker := storage.NewTreeChunker(storage.NewChunkerParams())
-	key, _, err := chunker.Split(f, stat.Size(), nil)
+	dpa := storage.NewDPA(storage.NewMapChunkStore(), storage.NewDPAParams())
+	key, _, err := dpa.Store(f, stat.Size(), false)
 	if err != nil {
 		utils.Fatalf("%v\n", err)
 	} else {

swarm/api/api.go

@@ -243,7 +243,7 @@ func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader {
 
 func (self *Api) Store(data io.Reader, size int64) (key storage.Key, wait func(), err error) {
 	log.Debug("api.store", "size", size)
-	return self.dpa.Store(data, size)
+	return self.dpa.Store(data, size, false)
 }
 
 type ErrResolve error
@@ -286,14 +286,14 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
 func (self *Api) Put(content, contentType string) (k storage.Key, wait func(), err error) {
 	apiPutCount.Inc(1)
 	r := strings.NewReader(content)
-	key, waitContent, err := self.dpa.Store(r, int64(len(content)))
+	key, waitContent, err := self.dpa.Store(r, int64(len(content)), false)
 	if err != nil {
 		apiPutFail.Inc(1)
 		return nil, nil, err
 	}
 	manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
 	r = strings.NewReader(manifest)
-	key, waitManifest, err := self.dpa.Store(r, int64(len(manifest)))
+	key, waitManifest, err := self.dpa.Store(r, int64(len(manifest)), false)
 	if err != nil {
 		apiPutFail.Inc(1)
 		return nil, nil, err

swarm/api/api_test.go

@@ -43,9 +43,7 @@ func testApi(t *testing.T, f func(*Api)) {
 		return
 	}
 	api := NewApi(dpa, nil, nil)
-	dpa.Start()
 	f(api)
-	dpa.Stop()
 }
 
 type testResponse struct {

swarm/api/config.go

@@ -43,7 +43,7 @@ const (
 type Config struct {
 	// serialised/persisted fields
 	*storage.StoreParams
-	*storage.ChunkerParams
+	*storage.DPAParams
 	*network.HiveParams
 	Swap *swap.SwapParams
 	//*network.SyncParams
@@ -72,9 +72,9 @@ type Config struct {
 func NewConfig() (self *Config) {
 
 	self = &Config{
-		StoreParams:   storage.NewDefaultStoreParams(),
-		ChunkerParams: storage.NewChunkerParams(),
-		HiveParams:    network.NewHiveParams(),
+		StoreParams: storage.NewDefaultStoreParams(),
+		DPAParams:   storage.NewDPAParams(),
+		HiveParams:  network.NewHiveParams(),
 		//SyncParams:    network.NewDefaultSyncParams(),
 		Swap:            swap.NewDefaultSwapParams(),
 		ListenAddr:      DefaultHTTPListenAddr,

swarm/api/filesystem.go

@@ -114,7 +114,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
 				stat, _ := f.Stat()
 				var hash storage.Key
 				var wait func()
-				hash, wait, err = self.api.dpa.Store(f, stat.Size())
+				hash, wait, err = self.api.dpa.Store(f, stat.Size(), false)
 				if hash != nil {
 					list[i].Hash = hash.Hex()
 				}

swarm/api/http/server_test.go

@@ -246,7 +246,7 @@ func TestBzzGetPath(t *testing.T) {
 	for i, mf := range testmanifest {
 		reader[i] = bytes.NewReader([]byte(mf))
 		var wait func()
-		key[i], wait, err = srv.Dpa.Store(reader[i], int64(len(mf)))
+		key[i], wait, err = srv.Dpa.Store(reader[i], int64(len(mf)), false)
 		if err != nil {
 			t.Fatal(err)
 		}

swarm/api/manifest.go

@@ -371,7 +371,7 @@ func (self *manifestTrie) recalcAndStore() error {
 	}
 
 	sr := bytes.NewReader(manifest)
-	key, wait, err2 := self.dpa.Store(sr, int64(len(manifest)))
+	key, wait, err2 := self.dpa.Store(sr, int64(len(manifest)), false)
 	wait()
 	self.hash = key
 	return err2

swarm/fuse/swarmfs_test.go

@@ -813,8 +813,6 @@ func TestFUSE(t *testing.T) {
 		t.Fatal(err)
 	}
 	ta := &testAPI{api: api.NewApi(dpa, nil, nil)}
-	dpa.Start()
-	defer dpa.Stop()
 
 	t.Run("mountListAndUmount", ta.mountListAndUnmount)
 	t.Run("maxMounts", ta.maxMounts)

swarm/network/stream/common_test.go

@@ -86,7 +86,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	go func() {
 		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
 	}()
-	dpa := storage.NewDPA(storage.NewNetStore(store, nil), storage.NewChunkerParams())
+	dpa := storage.NewDPA(storage.NewNetStore(store, nil), storage.NewDPAParams())
 	return &TestRegistry{Registry: r, dpa: dpa}, nil
 }
 
@@ -207,12 +207,10 @@ func (r *TestRegistry) ReadAll(hash common.Hash) (int64, error) {
 }
 
 func (r *TestRegistry) Start(server *p2p.Server) error {
-	r.dpa.Start()
 	return r.Registry.Start(server)
 }
 
 func (r *TestRegistry) Stop() error {
-	r.dpa.Stop()
 	return r.Registry.Stop()
 }
 

swarm/network/stream/delivery_test.go

@@ -341,13 +341,11 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 	}
 
 	// here we distribute chunks of a random file into Stores of nodes 1 to nodes
-	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
-	rrdpa.Start()
+	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewDPAParams())
 	size := chunkCount * chunkSize
-	fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size), false)
 	// wait until all chunks stored
 	wait()
-	defer rrdpa.Stop()
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -395,11 +393,9 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			return delivery.RequestFromPeers(chunk.Key[:], skipCheck)
 		}
 		netStore := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), retrieveFunc)
-		dpa := storage.NewDPA(netStore, storage.NewChunkerParams())
-		dpa.Start()
+		dpa := storage.NewDPA(netStore, storage.NewDPAParams())
 
 		go func() {
-			defer dpa.Stop()
 			// start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks
 			// we must wait for the peer connections to have started before requesting
 			n, err := readAll(dpa, fileHash)
@@ -518,9 +514,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 	waitPeerErrC = make(chan error)
 
 	// create a dpa for the last node in the chain which we are gonna write to
-	remoteDpa := storage.NewDPA(sim.Stores[nodes-1], storage.NewChunkerParams())
-	remoteDpa.Start()
-	defer remoteDpa.Stop()
+	remoteDpa := storage.NewDPA(sim.Stores[nodes-1], storage.NewDPAParams())
 
 	// channel to signal simulation initialisation with action call complete
 	// or node disconnections
@@ -614,7 +608,7 @@ Loop:
 		hashes := make([]storage.Key, chunkCount)
 		for i := 0; i < chunkCount; i++ {
 			// create actual size real chunks
-			hash, wait, err := remoteDpa.Store(io.LimitReader(crand.Reader, int64(chunkSize)), int64(chunkSize))
+			hash, wait, err := remoteDpa.Store(io.LimitReader(crand.Reader, int64(chunkSize)), int64(chunkSize), false)
 			// wait until all chunks stored
 			wait()
 			if err != nil {

swarm/network/stream/intervals_test.go

@@ -105,12 +105,10 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 		return 1
 	}
 
-	dpa := storage.NewDPA(sim.Stores[0], storage.NewChunkerParams())
-	dpa.Start()
+	dpa := storage.NewDPA(sim.Stores[0], storage.NewDPAParams())
 	size := chunkCount * chunkSize
-	_, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	_, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size), false)
 	wait()
-	defer dpa.Stop()
 	if err != nil {
 		t.Fatal(err)
 	}

swarm/network/stream/stream.go

@@ -93,7 +93,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
 		return NewSwarmChunkServer(delivery.db), nil
 	})
 	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ string, _ bool) (Client, error) {
-		return NewSwarmSyncerClient(p, delivery.db, nil)
+		return NewSwarmSyncerClient(p, delivery.db)
 	})
 	RegisterSwarmSyncerServer(streamer, db)
 	RegisterSwarmSyncerClient(streamer, db)

swarm/network/stream/syncer.go

@@ -17,10 +17,6 @@
 package stream
 
 import (
-	"bytes"
-	"errors"
-	"fmt"
-	"io"
 	"math"
 	"strconv"
 	"time"
@@ -138,17 +134,16 @@ type SwarmSyncerClient struct {
 	retrieveC     chan *storage.Chunk
 	storeC        chan *storage.Chunk
 	db            *storage.DBAPI
-	chunker       storage.Chunker
-	currentRoot   storage.Key
-	requestFunc   func(chunk *storage.Chunk)
-	end, start    uint64
+	// chunker       storage.Chunker
+	currentRoot storage.Key
+	requestFunc func(chunk *storage.Chunk)
+	end, start  uint64
 }
 
 // NewSwarmSyncerClient is a contructor for provable data exchange syncer
-func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (*SwarmSyncerClient, error) {
+func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI) (*SwarmSyncerClient, error) {
 	return &SwarmSyncerClient{
-		db:      db,
-		chunker: chunker,
+		db: db,
 	}, nil
 }
 
@@ -192,7 +187,7 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (
 // to handle incoming sync streams
 func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
 	streamer.RegisterClientFunc("SYNC", func(p *Peer, _ string, love bool) (Client, error) {
-		return NewSwarmSyncerClient(p, db, nil)
+		return NewSwarmSyncerClient(p, db)
 	})
 }
 
@@ -211,37 +206,39 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
 
 // BatchDone
 func (s *SwarmSyncerClient) BatchDone(stream Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
-	if s.chunker != nil {
-		return func() (*TakeoverProof, error) { return s.TakeoverProof(stream, from, hashes, root) }
-	}
+	// TODO: reenable this with putter/getter refactored code
+	// if s.chunker != nil {
+	// 	return func() (*TakeoverProof, error) { return s.TakeoverProof(stream, from, hashes, root) }
+	// }
 	return nil
 }
 
 func (s *SwarmSyncerClient) TakeoverProof(stream Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
 	// for provable syncer currentRoot is non-zero length
-	if s.chunker != nil {
-		if from > s.sessionAt { // for live syncing currentRoot is always updated
-			//expRoot, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC, s.storeC)
-			expRoot, _, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC)
-			if err != nil {
-				return nil, err
-			}
-			if !bytes.Equal(root, expRoot) {
-				return nil, fmt.Errorf("HandoverProof mismatch")
-			}
-			s.currentRoot = root
-		} else {
-			expHashes := make([]byte, len(hashes))
-			_, err := s.sessionReader.ReadAt(expHashes, int64(s.end*HashSize))
-			if err != nil && err != io.EOF {
-				return nil, err
-			}
-			if !bytes.Equal(expHashes, hashes) {
-				return nil, errors.New("invalid proof")
-			}
-		}
-		return nil, nil
-	}
+	// TODO: reenable this with putter/getter
+	// if s.chunker != nil {
+	// 	if from > s.sessionAt { // for live syncing currentRoot is always updated
+	// 		//expRoot, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC, s.storeC)
+	// 		expRoot, _, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC)
+	// 		if err != nil {
+	// 			return nil, err
+	// 		}
+	// 		if !bytes.Equal(root, expRoot) {
+	// 			return nil, fmt.Errorf("HandoverProof mismatch")
+	// 		}
+	// 		s.currentRoot = root
+	// 	} else {
+	// 		expHashes := make([]byte, len(hashes))
+	// 		_, err := s.sessionReader.ReadAt(expHashes, int64(s.end*HashSize))
+	// 		if err != nil && err != io.EOF {
+	// 			return nil, err
+	// 		}
+	// 		if !bytes.Equal(expHashes, hashes) {
+	// 			return nil, errors.New("invalid proof")
+	// 		}
+	// 	}
+	// 	return nil, nil
+	// }
 	s.end += uint64(len(hashes)) / HashSize
 	takeover := &Takeover{
 		Stream: stream,

swarm/network/stream/syncer_test.go

@@ -94,13 +94,11 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 	waitPeerErrC = make(chan error)
 
 	// here we distribute chunks of a random file into stores 1...nodes
-	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
-	rrdpa.Start()
+	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewDPAParams())
 	size := chunkCount * chunkSize
-	_, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	_, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size), false)
 	// need to wait cos we then immediately collect the relevant bin content
 	wait()
-	defer rrdpa.Stop()
 	if err != nil {
 		t.Fatal(err.Error())
 	}

swarm/pss/pss.go

@@ -775,7 +775,7 @@ func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool {
 // DPA storage handler for message cache
 func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
 	buf := bytes.NewReader(msg.serialize())
-	key, _, err := self.dpa.Store(buf, int64(buf.Len()))
+	key, _, err := self.dpa.Store(buf, int64(buf.Len()), false)
 	if err != nil {
 		log.Warn("Could not store in swarm", "err", err)
 		return pssDigest{}, err