Merge pull request #1787 from mysteriumnetwork/fix-proposal-panic

Exclude unsupported proposals from message broker

core/discovery/brokerdiscovery/repository.go

@@ -104,6 +104,10 @@ func (r *Repository) Stop() {
 }
 
 func (r *Repository) proposalRegisterMessage(message registerMessage) error {
+	if !message.Proposal.IsSupported() {
+		return nil
+	}
+
 	r.storage.AddProposal(message.Proposal)
 
 	r.watchdogLock.Lock()
@@ -124,6 +128,10 @@ func (r *Repository) proposalUnregisterMessage(message unregisterMessage) error
 }
 
 func (r *Repository) proposalPingMessage(message pingMessage) error {
+	if !message.Proposal.IsSupported() {
+		return nil
+	}
+
 	r.storage.AddProposal(message.Proposal)
 
 	r.watchdogLock.Lock()

core/discovery/brokerdiscovery/repository_test.go

@@ -18,27 +18,53 @@
 package brokerdiscovery
 
 import (
+	"encoding/json"
 	"testing"
 	"time"
 
 	"github.com/mysteriumnetwork/node/communication/nats"
 	"github.com/mysteriumnetwork/node/eventbus"
 	"github.com/mysteriumnetwork/node/market"
+	"github.com/mysteriumnetwork/node/money"
 	"github.com/stretchr/testify/assert"
 )
 
+func init() {
+	market.RegisterServiceDefinitionUnserializer(
+		"mock_service",
+		func(rawDefinition *json.RawMessage) (market.ServiceDefinition, error) {
+			return mockServiceDefinition{}, nil
+		},
+	)
+	market.RegisterPaymentMethodUnserializer(
+		"mock_payment",
+		func(rawDefinition *json.RawMessage) (market.PaymentMethod, error) {
+			return mockPaymentMethod{}, nil
+		},
+	)
+	market.RegisterContactUnserializer("mock_contact",
+		func(rawMessage *json.RawMessage) (market.ContactDefinition, error) {
+			return mockContact{}, nil
+		},
+	)
+}
+
 var (
 	proposalFirst = market.ServiceProposal{
 		ProviderID:        "0x1",
-		ServiceDefinition: market.UnsupportedServiceDefinition{},
-		PaymentMethod:     market.UnsupportedPaymentMethod{},
-		ProviderContacts:  []market.Contact{},
+		ServiceType:       "mock_service",
+		ServiceDefinition: mockServiceDefinition{},
+		PaymentMethodType: "mock_payment",
+		PaymentMethod:     mockPaymentMethod{},
+		ProviderContacts:  []market.Contact{market.Contact{Type: "mock_contact", Definition: mockContact{}}},
 	}
 	proposalSecond = market.ServiceProposal{
 		ProviderID:        "0x2",
-		ServiceDefinition: market.UnsupportedServiceDefinition{},
-		PaymentMethod:     market.UnsupportedPaymentMethod{},
-		ProviderContacts:  []market.Contact{},
+		ServiceType:       "mock_service",
+		ServiceDefinition: mockServiceDefinition{},
+		PaymentMethodType: "mock_payment",
+		PaymentMethod:     mockPaymentMethod{},
+		ProviderContacts:  []market.Contact{market.Contact{Type: "mock_contact", Definition: mockContact{}}},
 	}
 )
 
@@ -52,18 +78,29 @@ func Test_Subscriber_StartSyncsNewProposals(t *testing.T) {
 	assert.NoError(t, err)
 
 	proposalRegister(connection, `{
-		"proposal": {"provider_id": "0x1"}
+		"proposal": {"provider_id": "0x1", "service_type": "mock_service", "payment_method_type": "mock_payment", "provider_contacts": [{"type":"mock_contact"}]}
 	}`)
 
 	assert.Eventually(t, proposalCountEquals(repo, 1), 2*time.Second, 1*time.Millisecond)
 	assert.Exactly(t, []market.ServiceProposal{proposalFirst}, repo.storage.Proposals())
+}
+
+func Test_Subscriber_SkipUnsupportedProposal(t *testing.T) {
+	connection := nats.StartConnectionMock()
+	defer connection.Close()
+
+	repo := NewRepository(connection, eventbus.New(), 10*time.Millisecond, 1*time.Millisecond)
+	err := repo.Start()
+	defer repo.Stop()
+	assert.NoError(t, err)
 
 	proposalRegister(connection, `{
-		"proposal": {"provider_id": "0x2"}
+		"proposal": {"provider_id": "0x1", "service_type": "unknown"}
 	}`)
 
-	assert.Eventually(t, proposalCountEquals(repo, 2), 2*time.Second, 1*time.Millisecond)
-	assert.Exactly(t, []market.ServiceProposal{proposalFirst, proposalSecond}, repo.storage.Proposals())
+	time.Sleep(10 * time.Millisecond)
+	assert.Len(t, repo.storage.Proposals(), 0)
+	assert.Exactly(t, []market.ServiceProposal{}, repo.storage.Proposals())
 }
 
 func Test_Subscriber_StartSyncsIdleProposals(t *testing.T) {
@@ -91,11 +128,11 @@ func Test_Subscriber_StartSyncsHealthyProposals(t *testing.T) {
 	assert.NoError(t, err)
 
 	proposalRegister(connection, `{
-		"proposal": {"provider_id": "0x1"}
+		"proposal": {"provider_id": "0x1", "service_type": "mock_service", "payment_method_type": "mock_payment", "provider_contacts": [{"type":"mock_contact"}]}
 	}`)
 
 	proposalPing(connection, `{
-		"proposal": {"provider_id": "0x1"}
+		"proposal": {"provider_id": "0x1", "service_type": "mock_service", "payment_method_type": "mock_payment", "provider_contacts": [{"type":"mock_contact"}]}
 	}`)
 
 	assert.Eventually(t, proposalCountEquals(repo, 1), 2*time.Second, 1*time.Millisecond)
@@ -113,7 +150,7 @@ func Test_Subscriber_StartSyncsStoppedProposals(t *testing.T) {
 	assert.NoError(t, err)
 
 	proposalUnregister(connection, `{
-		"proposal": {"provider_id": "0x1"}
+		"proposal": {"provider_id": "0x1", "service_type": "mock_service", "payment_method_type": "mock_payment", "provider_contacts": [{"type":"mock_contact"}]}
 	}`)
 
 	assert.Eventually(t, proposalCountEquals(repo, 1), 2*time.Second, 1*time.Millisecond)
@@ -146,3 +183,29 @@ func proposalCountEquals(subscriber *Repository, count int) func() bool {
 		return len(subscriber.storage.Proposals()) == count
 	}
 }
+
+type mockServiceDefinition struct {
+}
+
+func (service mockServiceDefinition) GetLocation() market.Location {
+	return market.Location{}
+}
+
+type mockPaymentMethod struct {
+}
+
+func (method mockPaymentMethod) GetPrice() money.Money {
+	return money.Money{}
+}
+
+func (method mockPaymentMethod) GetType() string {
+	return "mock"
+}
+
+func (method mockPaymentMethod) GetRate() market.PaymentRate {
+	return market.PaymentRate{
+		PerTime: time.Minute,
+	}
+}
+
+type mockContact struct{}

docker-compose.e2e-basic.yml

@@ -89,7 +89,7 @@ services:
       - ./e2e/blockchain/keystore:/keystore
 
   accountant:
-    image: mysteriumnetwork/accountant:latest
+    image: mysteriumnetwork/accountant:0.0.2
     environment:
       PORT: 8889
     expose:

docker-compose.e2e-compatibility.yml

@@ -89,7 +89,7 @@ services:
       - ./e2e/blockchain/keystore:/keystore
 
   accountant:
-    image: mysteriumnetwork/accountant:latest
+    image: mysteriumnetwork/accountant:0.0.2
     environment:
       PORT: 8889
     expose:

docker-compose.e2e-traversal.yml

@@ -178,7 +178,7 @@ services:
         ipv4_address: 172.31.0.203
 
   accountant:
-    image: mysteriumnetwork/accountant:latest
+    image: mysteriumnetwork/accountant:0.0.2
     environment:
       PORT: 8889
     expose:

docker-compose.localnet.yml

@@ -178,7 +178,7 @@ services:
         ipv4_address: 172.31.0.203
 
   accountant:
-    image: mysteriumnetwork/accountant:latest
+    image: mysteriumnetwork/accountant:0.0.2
     environment:
       PORT: 8889
     expose:

market/payment_method.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/mysteriumnetwork/node/money"
+	"github.com/rs/zerolog/log"
 )
 
 // PaymentMethod is a interface for all types of payment methods
@@ -42,22 +43,25 @@ type PaymentRate struct {
 type UnsupportedPaymentMethod struct {
 }
 
-// GetPrice always panics for UnsupportedPaymentMethod and should not be called
+// GetPrice should not be called
 func (UnsupportedPaymentMethod) GetPrice() money.Money {
 	//this should never be called
-	panic("not supported")
+	log.Error().Msg("Unsupported proposal GetPrice should not be called")
+	return money.Money{}
 }
 
-// GetType always panics
+// GetType should not be called
 func (UnsupportedPaymentMethod) GetType() string {
 	//this should never be called
-	panic("not supported")
+	log.Error().Msg("Unsupported proposal GetType should not be called")
+	return ""
 }
 
-// GetRate always panics
+// GetRate should not be called
 func (UnsupportedPaymentMethod) GetRate() PaymentRate {
 	//this should never be called
-	panic("not supported")
+	log.Error().Msg("Unsupported proposal GetRate should not be called")
+	return PaymentRate{}
 }
 
 var _ PaymentMethod = UnsupportedPaymentMethod{}

market/service_definition.go

@@ -17,7 +17,11 @@
 
 package market
 
-import "encoding/json"
+import (
+	"encoding/json"
+
+	"github.com/rs/zerolog/log"
+)
 
 // ServiceDefinition interface is interface for all service definition types
 type ServiceDefinition interface {
@@ -28,10 +32,11 @@ type ServiceDefinition interface {
 type UnsupportedServiceDefinition struct {
 }
 
-// GetLocation always panics on unsupported service types
+// GetLocation should not be called for unsupported service types
 func (UnsupportedServiceDefinition) GetLocation() Location {
 	//no location available - should never be called
-	panic("not supported")
+	log.Error().Msg("Unsupported proposal GetLocation should not be called")
+	return Location{}
 }
 
 var _ ServiceDefinition = UnsupportedServiceDefinition{}