Merge pull request #641 from pastelnetwork/PSL-939_disableP2PKey

disable p2p key

Author: mateeullahmalik

dupedetection/ddclient/client.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	defaultConnectTimeout = 1000 * time.Second
+	defaultConnectTimeout = 60 * time.Second
 )
 
 type client struct{}
@@ -29,10 +29,7 @@ func (cl *client) Connect(ctx context.Context, address string) (*clientConn, err
 		//lint:ignore SA1019 we want to ignore this for now
 		grpc.WithInsecure(),
 		grpc.WithBlock(),
-		grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name), grpc.MaxCallRecvMsgSize(35000000),
-
-	
-	),
+		grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name), grpc.MaxCallRecvMsgSize(35000000)),
 	)
 	if err != nil {
 		return nil, errors.Errorf("fail to dial: %w", err).WithField("address", address)

dupedetection/ddclient/ddclient.go

@@ -21,7 +21,8 @@ import (
 )
 
 const (
-	logPrefix = "ddClient"
+	logPrefix                   = "ddClient"
+	defaultDDServiceCallTimeout = 26 * time.Minute
 )
 
 // DDServerClient contains methods for request services from dd-server service.
@@ -90,9 +91,13 @@ func (ddClient *ddServerClientImpl) callImageRarenessScore(ctx context.Context,
 	// remove file after use
 	defer os.Remove(inputPath)
 
-	res, err := client.ImageRarenessScore(context.Background(), &req, grpc.MaxCallRecvMsgSize(35000000))
+	// Limits the dial timeout, prevent got stuck too long
+	dialCtx, dcancel := context.WithTimeout(ctx, defaultDDServiceCallTimeout)
+	defer dcancel()
+
+	res, err := client.ImageRarenessScore(dialCtx, &req, grpc.MaxCallRecvMsgSize(35000000))
 	if err != nil {
-		return nil, errors.Errorf("Error calling image rareness score: %w", err)
+		return nil, errors.Errorf("Error calling image rareness score, dd-serivce returned error: %w", err)
 	}
 
 	bytes, _ := json.Marshal(res)

p2p/kademlia/dht.go

@@ -24,10 +24,12 @@ import (
 )
 
 var (
-	defaultNetworkAddr = "0.0.0.0"
-	defaultNetworkPort = 4445
-	defaultRefreshTime = time.Second * 3600
-	defaultPingTime    = time.Second * 10
+	defaultNetworkAddr                   = "0.0.0.0"
+	defaultNetworkPort                   = 4445
+	defaultRefreshTime                   = time.Second * 3600
+	defaultPingTime                      = time.Second * 10
+	defaultCleanupInterval               = time.Minute * 2
+	defaultDisabledKeyExpirationInterval = time.Minute * 30
 )
 
 const maxIterations = 5
@@ -38,6 +40,7 @@ type DHT struct {
 	options              *Options         // the options of DHT
 	network              *Network         // the network of DHT
 	store                Store            // the storage of DHT
+	metaStore            MetaStore        // the meta storage of DHT
 	done                 chan struct{}    // distributed hash table is done
 	cache                storage.KeyValue // store bad bootstrap addresses
 	pastelClient         pastel.Client
@@ -73,7 +76,7 @@ type Options struct {
 }
 
 // NewDHT returns a new DHT node
-func NewDHT(ctx context.Context, store Store, pc pastel.Client, secInfo *alts.SecInfo, options *Options) (*DHT, error) {
+func NewDHT(ctx context.Context, store Store, metaStore MetaStore, pc pastel.Client, secInfo *alts.SecInfo, options *Options) (*DHT, error) {
 	// validate the options, if it's invalid, set them to default value
 	if options.IP == "" {
 		options.IP = defaultNetworkAddr
@@ -92,6 +95,7 @@ func NewDHT(ctx context.Context, store Store, pc pastel.Client, secInfo *alts.Se
 	}
 
 	s := &DHT{
+		metaStore:            metaStore,
 		store:                store,
 		options:              options,
 		pastelClient:         pc,
@@ -165,6 +169,7 @@ func (s *DHT) Start(ctx context.Context) error {
 	}
 
 	go s.StartReplicationWorker(ctx)
+	go s.startDisabledKeysCleanupWorker(ctx)
 
 	return nil
 }
@@ -278,6 +283,12 @@ func (s *DHT) Retrieve(ctx context.Context, key string, localOnly ...bool) ([]by
 	}
 
 	dbKey := hex.EncodeToString(decoded)
+	if s.metaStore != nil {
+		if err := s.metaStore.Retrieve(ctx, dbKey); err == nil {
+			return nil, fmt.Errorf("key is disabled: %v", key)
+		}
+	}
+
 	// retrieve the key/value from local storage
 	value, err := s.store.Retrieve(ctx, decoded)
 	if err == nil && len(value) > 0 {
@@ -871,3 +882,40 @@ func (s *DHT) storeToAlphaNodes(ctx context.Context, nl *NodeList, data []byte,
 
 	return fmt.Errorf("store data to alpha nodes failed, only %d nodes stored", finalStoreCount)
 }
+
+func (s *DHT) startDisabledKeysCleanupWorker(ctx context.Context) error {
+	log.P2P().WithContext(ctx).Info("disabled keys cleanup worker started")
+
+	for {
+		select {
+		case <-time.After(defaultCleanupInterval):
+			s.cleanupDisabledKeys(ctx)
+		case <-ctx.Done():
+			log.P2P().WithContext(ctx).Error("closing disabled keys cleanup worker")
+			return nil
+		}
+	}
+}
+
+func (s *DHT) cleanupDisabledKeys(ctx context.Context) error {
+	if s.metaStore == nil {
+		return nil
+	}
+
+	from := time.Now().Add(-1 * defaultDisabledKeyExpirationInterval)
+	disabledKeys, err := s.metaStore.GetDisabledKeys(from)
+	if err != nil {
+		return errors.Errorf("get disabled keys: %w", err)
+	}
+
+	for i := 0; i < len(disabledKeys); i++ {
+		dec, err := hex.DecodeString(disabledKeys[i].Key)
+		if err != nil {
+			log.P2P().WithContext(ctx).WithError(err).Error("decode disabled key failed")
+			continue
+		}
+		s.metaStore.Delete(ctx, dec)
+	}
+
+	return nil
+}

p2p/kademlia/dht_test.go

@@ -226,7 +226,7 @@ func (ts *testSuite) newDHTNodeWithMemStore(ctx context.Context, port int, nodes
 	// }
 	// transportCredentials := credentials.NewClientCreds(fakePastelClient, secInfo)
 
-	dht, err := NewDHT(ctx, ts.memStore, pastelClientMock, secInfo, options)
+	dht, err := NewDHT(ctx, ts.memStore, nil, pastelClientMock, secInfo, options)
 	if err != nil {
 		return nil, errors.Errorf("new dht: %w", err)
 	}
@@ -270,7 +270,7 @@ func (ts *testSuite) newDHTNodeWithDBStore(ctx context.Context, port int, nodes
 		pub:    TestPub,
 	}
 
-	dht, err := NewDHT(ctx, ts.dbStore, fakePastelClient, secInfo, options)
+	dht, err := NewDHT(ctx, ts.dbStore, nil, fakePastelClient, secInfo, options)
 	if err != nil {
 		return nil, errors.Errorf("new dht: %w", err)
 	}

p2p/kademlia/domain/domain.go

@@ -27,3 +27,12 @@ type ToRepKey struct {
 	ID        string    `json:"id"`
 	Attempts  int       `json:"attempts"`
 }
+
+// DisabledKey is a disabled key
+type DisabledKey struct {
+	Key       string    `json:"key"`
+	CreatedAt time.Time `json:"createdAt"`
+}
+
+// DisabledKeys is the list for disabled keys
+type DisabledKeys []DisabledKey

p2p/kademlia/replication.go

@@ -55,7 +55,8 @@ func (s *DHT) StartReplicationWorker(ctx context.Context) error {
 	for {
 		select {
 		case <-time.After(defaultReplicationInterval):
-			s.Replicate(ctx)
+			//s.Replicate(ctx)
+			log.P2P().WithContext(ctx).Info("replication worker currently disabled")
 		case <-ctx.Done():
 			log.P2P().WithContext(ctx).Error("closing replication worker")
 			return nil

p2p/kademlia/store.go

@@ -73,3 +73,21 @@ type Store interface {
 
 	IncrementAttempts(keys []string) error
 }
+
+// MetaStore is the interface for implementing the storage mechanism for the DHT
+type MetaStore interface {
+	// Delete a key pair from the store
+	Delete(_ context.Context, key []byte)
+
+	// Retrieve the local key from store
+	Retrieve(_ context.Context, key string) error
+
+	// Close the store
+	Close(ctx context.Context)
+
+	// Store a key pair for the local node with the replication
+	Store(ctx context.Context, key []byte) error
+
+	// GetDisabledKeys returns all disabled keys
+	GetDisabledKeys(from time.Time) (retKeys domain.DisabledKeys, err error)
+}