api, cmd, prod, pss, swarm: get publisher and query feed for chunk repair request

chunk/chunk.go

@@ -283,7 +283,7 @@ type Validator interface {
 // with validators check.
 type ValidatorStore struct {
 	Store
-	deliverCallback func(Chunk) // callback hook used to validate chunk
+	deliverCallback func(Chunk) // callback func to be invoked for validated chunks
 	validators      []Validator
 }
 

prod/prod.go

@@ -18,44 +18,97 @@ package prod
 
 import (
 	"context"
+	"encoding/hex"
+	"errors"
+	"fmt"
 
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethersphere/swarm/chunk"
 	"github.com/ethersphere/swarm/pss"
 	"github.com/ethersphere/swarm/pss/trojan"
+	"github.com/ethersphere/swarm/storage/feed"
+	"github.com/ethersphere/swarm/storage/feed/lookup"
 )
 
-// RecoveryHook defines code to be executed upon trigger of failed to be retrieved chunks
-type RecoveryHook func(ctx context.Context, chunkAddress chunk.Address) error
+// ErrPublisher is returned when the publisher string cannot be decoded into bytes
+var ErrPublisher = errors.New("failed to decode publisher")
 
-// sender is the function call for sending trojan chunks
+// ErrPubKey is returned when the publisher bytes cannot be decompressed as a public key
+var ErrPubKey = errors.New("failed to decompress public key")
+
+// ErrFeedLookup is returned when the recovery feed cannot be successefully looked up
+var ErrFeedLookup = errors.New("failed to look up recovery feed")
+
+// ErrFeedContent is returned when there is a failure to retrieve content from the recovery feed
+var ErrFeedContent = errors.New("failed to get content for recovery feed")
+
+// RecoveryHook defines code to be executed upon failing to retrieve pinned chunks
+type RecoveryHook func(ctx context.Context, chunkAddress chunk.Address, publisher string) error
+
+// sender is the function type for sending trojan chunks
 type sender func(ctx context.Context, targets [][]byte, topic trojan.Topic, payload []byte) (*pss.Monitor, error)
 
 // NewRecoveryHook returns a new RecoveryHook with the sender function defined
-func NewRecoveryHook(send sender) RecoveryHook {
-	return func(ctx context.Context, chunkAddress chunk.Address) error {
-		targets, err := getPinners(chunkAddress)
+func NewRecoveryHook(send sender, handler feed.GenericHandler) RecoveryHook {
+	return func(ctx context.Context, chunkAddress chunk.Address, publisher string) error {
+		targets, err := getPinners(publisher, handler)
 		if err != nil {
 			return err
 		}
 		payload := chunkAddress
-		topic := trojan.NewTopic("RECOVERY")
 
 		// TODO: monitor return should
-		if _, err := send(ctx, targets, topic, payload); err != nil {
+		if _, err := send(ctx, targets, trojan.RecoveryTopic, payload); err != nil {
 			return err
 		}
 		return nil
 	}
 }
 
-// TODO: refactor this method to implement feed of target pinners
-// getPinners returns the specific target pinners for a corresponding chunk address
-func getPinners(chunkAddress chunk.Address) ([][]byte, error) {
-	//this should get the feed and return correct target of pinners
-	return [][]byte{
-		{57, 120},
-		{209, 156},
-		{156, 38},
-		{89, 19},
-		{22, 129}}, nil
+// getPinners returns the specific target pinners for a corresponding chunk
+func getPinners(publisher string, handler feed.GenericHandler) ([][]byte, error) {
+	// get feed user from publisher
+	publisherBytes, err := hex.DecodeString(publisher)
+	if err != nil {
+		return nil, ErrPublisher
+	}
+	pubKey, err := crypto.DecompressPubkey(publisherBytes)
+	if err != nil {
+		return nil, ErrPubKey
+	}
+	addr := crypto.PubkeyToAddress(*pubKey)
+
+	// get feed topic from trojan recovery topic
+	topic, err := feed.NewTopic(trojan.RecoveryTopicText, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	// read feed
+	fd := feed.Feed{
+		Topic: topic,
+		User:  addr,
+	}
+
+	query := feed.NewQueryLatest(&fd, lookup.NoClue)
+	// TODO: do we need this ctx.WithCancel? why?
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// TODO: not exactly sure what to do with the `feed.cacheEntry` value returned here
+	// TODO: in fact, not even sure if we need to call `Lookup` first before calling `GetContent`
+	_, err = handler.Lookup(ctx, query)
+	// feed can still be queried even if there are no updates
+	if err != nil && err.Error() != "no feed updates found" {
+		return nil, fmt.Errorf("%s : %s", ErrFeedLookup, err)
+	}
+
+	// TODO: how do we handle time-outs here?
+	_, content, err := handler.GetContent(&fd)
+	if err != nil {
+		return nil, fmt.Errorf("%s : %s", ErrFeedContent, err)
+	}
+
+	// TODO: transform content into actual list of targets
+	return [][]byte{content}, nil
 }

prod/prod_test.go

@@ -31,6 +31,7 @@ import (
 	psstest "github.com/ethersphere/swarm/pss/testing"
 	"github.com/ethersphere/swarm/pss/trojan"
 	"github.com/ethersphere/swarm/storage"
+	"github.com/ethersphere/swarm/storage/feed"
 )
 
 // TestRecoveryHook tests that NewRecoveryHook has been properly invoked
@@ -44,11 +45,16 @@ func TestRecoveryHook(t *testing.T) {
 		hookWasCalled = true
 		return nil, nil
 	}
+	testHandler := &feed.DummyHandler{}
 
 	// setup recovery hook with testHook
-	recoverFunc := NewRecoveryHook(testHook)
+	recoverFunc := NewRecoveryHook(testHook, testHandler)
 
-	recoverFunc(ctx, chunk.ZeroAddr)
+	testChunk := "aacca8d446af47ebcab582ca2188fa73dfa871eb0a35eda798f47d4f91a575e9"
+	testPublisher := "0226f213613e843a413ad35b40f193910d26eb35f00154afcde9ded57479a6224a"
+	if err := recoverFunc(ctx, chunk.Address([]byte(testChunk)), testPublisher); err != nil {
+		t.Fatal(err)
+	}
 
 	// verify the hook has been called correctly
 	if hookWasCalled != true {
@@ -85,8 +91,9 @@ func TestSenderCall(t *testing.T) {
 		hookWasCalled = true
 		return nil, nil
 	}
+	testHandler := &feed.DummyHandler{}
 
-	recoverFunc := NewRecoveryHook(testHook)
+	recoverFunc := NewRecoveryHook(testHook, testHandler)
 	netStore.WithRecoveryCallback(recoverFunc)
 
 	c := ctest.GenerateTestRandomChunk()

pss/pss_test.go

@@ -45,10 +45,9 @@ func TestTrojanChunkRetrieval(t *testing.T) {
 		{89, 19},
 		{22, 129}}
 	payload := []byte("RECOVERY CHUNK")
-	topic := trojan.NewTopic("RECOVERY")
 
 	// call Send to store trojan chunk in localstore
-	if _, err = pss.Send(ctx, targets, topic, payload); err != nil {
+	if _, err = pss.Send(ctx, targets, trojan.RecoveryTopic, payload); err != nil {
 		t.Fatal(err)
 	}
 
@@ -119,14 +118,13 @@ func TestPssMonitor(t *testing.T) {
 		{89, 19},
 		{22, 129}}
 	payload := []byte("RECOVERY CHUNK")
-	topic := trojan.NewTopic("RECOVERY")
 
 	var monitor *Monitor
 
 	pss := NewPss(localStore, tags)
 
 	// call Send to store trojan chunk in localstore
-	if monitor, err = pss.Send(ctx, targets, topic, payload); err != nil {
+	if monitor, err = pss.Send(ctx, targets, trojan.RecoveryTopic, payload); err != nil {
 		t.Fatal(err)
 	}
 

pss/trojan/message.go

@@ -45,6 +45,12 @@ const MaxPayloadSize = 4030
 
 var hashFunc = storage.MakeHashFunc(storage.BMTHash)
 
+// RecoveryTopicText is the string used to construct the RecoveryTopic var
+const RecoveryTopicText = "RECOVERY"
+
+// RecoveryTopic is the topic used for repairing globally pinned chunks
+var RecoveryTopic = NewTopic(RecoveryTopicText)
+
 // ErrPayloadTooBig is returned when a given payload for a Message type is longer than the maximum amount allowed
 var ErrPayloadTooBig = fmt.Errorf("message payload size cannot be greater than %d bytes", MaxPayloadSize)
 

storage/feed/handler.go

@@ -31,6 +31,13 @@ import (
 	"github.com/ethersphere/swarm/storage/feed/lookup"
 )
 
+// GenericHandler is an interface which specifies funcs any feeds handler should use
+type GenericHandler interface {
+	Lookup(ctx context.Context, query *Query) (*cacheEntry, error)
+	GetContent(feed *Feed) (storage.Address, []byte, error)
+}
+
+// Handler is the struct to be used as the API for feeds
 type Handler struct {
 	chunkStore *storage.NetStore
 	HashSize   int

storage/feed/testutil.go

@@ -17,6 +17,7 @@
 package feed
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"path/filepath"
@@ -32,15 +33,12 @@ const (
 	testDbDirName = "feeds"
 )
 
+// TestHandler is a stuct to be used for testing purposes
 type TestHandler struct {
 	*Handler
 }
 
-func (t *TestHandler) Close() {
-	t.chunkStore.Close()
-}
-
-// NewTestHandler creates Handler object to be used for testing purposes.
+// NewTestHandler creates a new TestHandler
 func NewTestHandler(datadir string, params *HandlerParams) (*TestHandler, error) {
 	path := filepath.Join(datadir, testDbDirName)
 	fh := NewHandler(params)
@@ -60,6 +58,7 @@ func NewTestHandler(datadir string, params *HandlerParams) (*TestHandler, error)
 	return &TestHandler{fh}, nil
 }
 
+// NewTestHandlerWithStore creates a new TestHandler with a set chunk store
 func NewTestHandlerWithStore(datadir string, db chunk.Store, params *HandlerParams) (*TestHandler, error) {
 	fh := NewHandler(params)
 	return newTestHandlerWithStore(fh, datadir, db, params)
@@ -75,3 +74,39 @@ func newTestHandlerWithStore(fh *Handler, datadir string, db chunk.Store, params
 	fh.SetStore(netStore)
 	return &TestHandler{fh}, nil
 }
+
+// Close closes the chunk store field for the test handler receiver
+func (t *TestHandler) Close() {
+	t.chunkStore.Close()
+}
+
+// DummyHandler is a test handler with dummy Lookup and GetContent funcs
+type DummyHandler struct {
+}
+
+// Lookup is the dummy func for DummyHandler structs
+func (d *DummyHandler) Lookup(ctx context.Context, query *Query) (*cacheEntry, error) {
+	return newDummyCacheEntry(), nil
+}
+
+// GetContent is the dummy func for DummyHandler structs
+func (d *DummyHandler) GetContent(feed *Feed) (storage.Address, []byte, error) {
+	cacheEntry := newDummyCacheEntry()
+	return cacheEntry.lastKey, cacheEntry.data, nil
+}
+
+// newDummyCacheEntry returns a dummy cacheEntry pointer to be used by dummy handlers
+func newDummyCacheEntry() *cacheEntry {
+	topic := Topic{0x1}             // dummy topic
+	key := chunk.Address([]byte{2}) // dummy byte
+	data := []byte{3}               // dummy data
+
+	request := NewFirstRequest(topic)
+	entry := &cacheEntry{}
+	entry.lastKey = key
+	entry.Update = request.Update
+
+	entry.Reader = bytes.NewReader(data)
+
+	return entry
+}

storage/netstore.go

@@ -99,7 +99,7 @@ type NetStore struct {
 	requestGroup     singleflight.Group
 	RemoteGet        RemoteGetFunc
 	logger           log.Logger
-	recoveryCallback func(ctx context.Context, chunkAddress chunk.Address) error // this is the function callback when a chunk failed to be retrieved
+	recoveryCallback func(ctx context.Context, chunkAddress chunk.Address, publisher string) error // this is the callback to be executed when a chunk fails to be retrieved
 }
 
 // NewNetStore creates a new NetStore using the provided chunk.Store and localID of the node.
@@ -167,7 +167,7 @@ func (n *NetStore) Close() error {
 }
 
 // WithRecoveryCallback allows injecting a callback func on the NetStore struct
-func (n *NetStore) WithRecoveryCallback(f func(ctx context.Context, chunkAddress chunk.Address) error) *NetStore {
+func (n *NetStore) WithRecoveryCallback(f func(ctx context.Context, chunkAddress chunk.Address, publisher string) error) *NetStore {
 	n.recoveryCallback = f
 	return n
 }
@@ -201,7 +201,8 @@ func (n *NetStore) Get(ctx context.Context, mode chunk.ModeGet, req *Request) (c
 				ch, err = n.RemoteFetch(ctx, req, fi)
 				if err != nil {
 					if n.recoveryCallback != nil {
-						n.recoveryCallback(ctx, ref)
+						// TODO: get actual publisher, dummy for now
+						n.recoveryCallback(ctx, ref, "0226f213613e843a413ad35b40f193910d26eb35f00154afcde9ded57479a6224a")
 						time.Sleep(500 * time.Millisecond) // TODO: view what the ideal timeout is
 						return n.RemoteFetch(ctx, req, fi)
 					}

swarm.go

@@ -291,8 +291,8 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 			chAddr := m.Payload
 			lstore.Set(context.Background(), chunk.ModeSetReUpload, chAddr)
 		}
-		self.pss.Register(trojan.NewTopic("RECOVERY"), repairFunc)
-		recoverFunc := prod.NewRecoveryHook(self.pss.Send)
+		self.pss.Register(trojan.RecoveryTopic, repairFunc)
+		recoverFunc := prod.NewRecoveryHook(self.pss.Send, feedsHandler)
 		self.netStore.WithRecoveryCallback(recoverFunc)
 	}