Localstore integration

swarm/network/stream/delivery.go

@@ -230,13 +230,21 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req int
 	switch r := req.(type) {
 	case *ChunkDeliveryMsgRetrieval:
 		msg = (*ChunkDeliveryMsg)(r)
-		// do not sync if peer that is sending us a chunk is closer to the chunk then we are
 		peerPO := chunk.Proximity(sp.ID().Bytes(), msg.Addr)
 		po := chunk.Proximity(d.kad.BaseAddr(), msg.Addr)
-		if peerPO > po {
-			mode = chunk.ModePutRequest
-		} else {
+		depth := d.kad.NeighbourhoodDepth()
+		// is the chunk within our area of responsibility?
+		if po < depth {
+			// chunks within the area of responsibility should always sync
+			// https://github.com/ethersphere/go-ethereum/pull/1282#discussion_r269406125
 			mode = chunk.ModePutSync
+		} else {
+			if peerPO > po {
+				// do not sync if peer that is sending us a chunk is closer to the chunk then we are
+				mode = chunk.ModePutRequest
+			} else {
+				mode = chunk.ModePutSync
+			}
 		}
 	case *ChunkDeliveryMsgSyncing:
 		msg = (*ChunkDeliveryMsg)(r)

swarm/storage/localstore/export_test.go

@@ -18,6 +18,7 @@ package localstore
 
 import (
 	"bytes"
+	"context"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/swarm/chunk"
@@ -36,7 +37,7 @@ func TestExportImport(t *testing.T) {
 	for i := 0; i < chunkCount; i++ {
 		ch := generateTestRandomChunk()
 
-		err := db1.NewPutter(chunk.ModePutUpload).Put(ch)
+		err := db1.Put(context.Background(), chunk.ModePutUpload, ch)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -65,11 +66,9 @@ func TestExportImport(t *testing.T) {
 		t.Errorf("got import count %v, want %v", c, wantChunksCount)
 	}
 
-	getter := db2.NewGetter(chunk.ModeGetRequest)
-
 	for a, want := range chunks {
 		addr := chunk.Address([]byte(a))
-		ch, err := getter.Get(addr)
+		ch, err := db2.Get(context.Background(), chunk.ModeGetRequest, addr)
 		if err != nil {
 			t.Fatal(err)
 		}

swarm/storage/localstore/gc_test.go

@@ -17,6 +17,7 @@
 package localstore
 
 import (
+	"context"
 	"io/ioutil"
 	"math/rand"
 	"os"
@@ -63,26 +64,23 @@ func testDB_collectGarbageWorker(t *testing.T) {
 	})()
 	defer cleanupFunc()
 
-	uploader := db.NewPutter(chunk.ModePutUpload)
-	syncer := db.NewSetter(chunk.ModeSetSync)
-
 	addrs := make([]chunk.Address, 0)
 
 	// upload random chunks
 	for i := 0; i < chunkCount; i++ {
-		chunk := generateTestRandomChunk()
+		ch := generateTestRandomChunk()
 
-		err := uploader.Put(chunk)
+		err := db.Put(context.Background(), chunk.ModePutUpload, ch)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		err = syncer.Set(chunk.Address())
+		err = db.Set(context.Background(), chunk.ModeSetSync, ch.Address())
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		addrs = append(addrs, chunk.Address())
+		addrs = append(addrs, ch.Address())
 	}
 
 	gcTarget := db.gcTarget()
@@ -110,15 +108,15 @@ func testDB_collectGarbageWorker(t *testing.T) {
 
 	// the first synced chunk should be removed
 	t.Run("get the first synced chunk", func(t *testing.T) {
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(addrs[0])
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, addrs[0])
 		if err != chunk.ErrChunkNotFound {
 			t.Errorf("got error %v, want %v", err, chunk.ErrChunkNotFound)
 		}
 	})
 
 	// last synced chunk should not be removed
 	t.Run("get most recent synced chunk", func(t *testing.T) {
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(addrs[len(addrs)-1])
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, addrs[len(addrs)-1])
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -134,9 +132,6 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 	})
 	defer cleanupFunc()
 
-	uploader := db.NewPutter(chunk.ModePutUpload)
-	syncer := db.NewSetter(chunk.ModeSetSync)
-
 	testHookCollectGarbageChan := make(chan uint64)
 	defer setTestHookCollectGarbage(func(collectedCount uint64) {
 		testHookCollectGarbageChan <- collectedCount
@@ -146,19 +141,19 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 
 	// upload random chunks just up to the capacity
 	for i := 0; i < int(db.capacity)-1; i++ {
-		chunk := generateTestRandomChunk()
+		ch := generateTestRandomChunk()
 
-		err := uploader.Put(chunk)
+		err := db.Put(context.Background(), chunk.ModePutUpload, ch)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		err = syncer.Set(chunk.Address())
+		err = db.Set(context.Background(), chunk.ModeSetSync, ch.Address())
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		addrs = append(addrs, chunk.Address())
+		addrs = append(addrs, ch.Address())
 	}
 
 	// set update gc test hook to signal when
@@ -172,7 +167,7 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 	// request the latest synced chunk
 	// to prioritize it in the gc index
 	// not to be collected
-	_, err := db.NewGetter(chunk.ModeGetRequest).Get(addrs[0])
+	_, err := db.Get(context.Background(), chunk.ModeGetRequest, addrs[0])
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -191,11 +186,11 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 	// upload and sync another chunk to trigger
 	// garbage collection
 	ch := generateTestRandomChunk()
-	err = uploader.Put(ch)
+	err = db.Put(context.Background(), chunk.ModePutUpload, ch)
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = syncer.Set(ch.Address())
+	err = db.Set(context.Background(), chunk.ModeSetSync, ch.Address())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -235,23 +230,23 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 
 	// requested chunk should not be removed
 	t.Run("get requested chunk", func(t *testing.T) {
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(addrs[0])
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, addrs[0])
 		if err != nil {
 			t.Fatal(err)
 		}
 	})
 
 	// the second synced chunk should be removed
 	t.Run("get gc-ed chunk", func(t *testing.T) {
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(addrs[1])
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, addrs[1])
 		if err != chunk.ErrChunkNotFound {
 			t.Errorf("got error %v, want %v", err, chunk.ErrChunkNotFound)
 		}
 	})
 
 	// last synced chunk should not be removed
 	t.Run("get most recent synced chunk", func(t *testing.T) {
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(addrs[len(addrs)-1])
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, addrs[len(addrs)-1])
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -275,20 +270,17 @@ func TestDB_gcSize(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	uploader := db.NewPutter(chunk.ModePutUpload)
-	syncer := db.NewSetter(chunk.ModeSetSync)
-
 	count := 100
 
 	for i := 0; i < count; i++ {
-		chunk := generateTestRandomChunk()
+		ch := generateTestRandomChunk()
 
-		err := uploader.Put(chunk)
+		err := db.Put(context.Background(), chunk.ModePutUpload, ch)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		err = syncer.Set(chunk.Address())
+		err = db.Set(context.Background(), chunk.ModeSetSync, ch.Address())
 		if err != nil {
 			t.Fatal(err)
 		}

swarm/storage/localstore/index_test.go

@@ -18,6 +18,7 @@ package localstore
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"math"
 	"math/rand"
@@ -39,23 +40,21 @@ func TestDB_pullIndex(t *testing.T) {
 	db, cleanupFunc := newTestDB(t, nil)
 	defer cleanupFunc()
 
-	uploader := db.NewPutter(chunk.ModePutUpload)
-
 	chunkCount := 50
 
 	chunks := make([]testIndexChunk, chunkCount)
 
 	// upload random chunks
 	for i := 0; i < chunkCount; i++ {
-		chunk := generateTestRandomChunk()
+		ch := generateTestRandomChunk()
 
-		err := uploader.Put(chunk)
+		err := db.Put(context.Background(), chunk.ModePutUpload, ch)
 		if err != nil {
 			t.Fatal(err)
 		}
 
 		chunks[i] = testIndexChunk{
-			Chunk: chunk,
+			Chunk: ch,
 			binID: uint64(i),
 		}
 	}
@@ -86,23 +85,21 @@ func TestDB_gcIndex(t *testing.T) {
 	db, cleanupFunc := newTestDB(t, nil)
 	defer cleanupFunc()
 
-	uploader := db.NewPutter(chunk.ModePutUpload)
-
 	chunkCount := 50
 
 	chunks := make([]testIndexChunk, chunkCount)
 
 	// upload random chunks
 	for i := 0; i < chunkCount; i++ {
-		chunk := generateTestRandomChunk()
+		ch := generateTestRandomChunk()
 
-		err := uploader.Put(chunk)
+		err := db.Put(context.Background(), chunk.ModePutUpload, ch)
 		if err != nil {
 			t.Fatal(err)
 		}
 
 		chunks[i] = testIndexChunk{
-			Chunk: chunk,
+			Chunk: ch,
 		}
 	}
 
@@ -124,7 +121,7 @@ func TestDB_gcIndex(t *testing.T) {
 	t.Run("request unsynced", func(t *testing.T) {
 		ch := chunks[1]
 
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(ch.Address())
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, ch.Address())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -141,7 +138,7 @@ func TestDB_gcIndex(t *testing.T) {
 	t.Run("sync one chunk", func(t *testing.T) {
 		ch := chunks[0]
 
-		err := db.NewSetter(chunk.ModeSetSync).Set(ch.Address())
+		err := db.Set(context.Background(), chunk.ModeSetSync, ch.Address())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -153,10 +150,8 @@ func TestDB_gcIndex(t *testing.T) {
 	})
 
 	t.Run("sync all chunks", func(t *testing.T) {
-		setter := db.NewSetter(chunk.ModeSetSync)
-
 		for i := range chunks {
-			err := setter.Set(chunks[i].Address())
+			err := db.Set(context.Background(), chunk.ModeSetSync, chunks[i].Address())
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -170,7 +165,7 @@ func TestDB_gcIndex(t *testing.T) {
 	t.Run("request one chunk", func(t *testing.T) {
 		i := 6
 
-		_, err := db.NewGetter(chunk.ModeGetRequest).Get(chunks[i].Address())
+		_, err := db.Get(context.Background(), chunk.ModeGetRequest, chunks[i].Address())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -188,14 +183,13 @@ func TestDB_gcIndex(t *testing.T) {
 	})
 
 	t.Run("random chunk request", func(t *testing.T) {
-		requester := db.NewGetter(chunk.ModeGetRequest)
 
 		rand.Shuffle(len(chunks), func(i, j int) {
 			chunks[i], chunks[j] = chunks[j], chunks[i]
 		})
 
-		for _, chunk := range chunks {
-			_, err := requester.Get(chunk.Address())
+		for _, ch := range chunks {
+			_, err := db.Get(context.Background(), chunk.ModeGetRequest, ch.Address())
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -211,7 +205,7 @@ func TestDB_gcIndex(t *testing.T) {
 	t.Run("remove one chunk", func(t *testing.T) {
 		i := 3
 
-		err := db.NewSetter(chunk.ModeSetRemove).Set(chunks[i].Address())
+		err := db.Set(context.Background(), chunk.ModeSetRemove, chunks[i].Address())
 		if err != nil {
 			t.Fatal(err)
 		}