Add 20MB of leeway for consumer

cmd/di.go

@@ -520,6 +520,7 @@ func (di *Dependencies) bootstrapNodeComponents(nodeOptions node.Options, tequil
 			nodeOptions.Transactor.RegistryAddress,
 			di.EventBus,
 			consumerDataGetter,
+			nodeOptions.Payments.ConsumerDataLeewayMegabytes,
 		),
 		di.ConnectionRegistry.CreateConnection,
 		di.EventBus,

config/flags_payments.go

@@ -84,6 +84,12 @@ var (
 		Usage: "Sets the minimum price of the service per gb. All proposals with a below above this bound will be filtered out and not visible.",
 		Value: 0,
 	}
+	// FlagPaymentsConsumerDataLeewayMegabytes sets the data amount the consumer agrees to pay before establishing a session
+	FlagPaymentsConsumerDataLeewayMegabytes = cli.Uint64Flag{
+		Name:  "payments.consumer.data-leeway-megabytes",
+		Usage: "sets the data amount the consumer agrees to pay before establishing a session",
+		Value: 20,
+	}
 )
 
 // RegisterFlagsPayments function register payments flags to flag list.
@@ -100,6 +106,7 @@ func RegisterFlagsPayments(flags *[]cli.Flag) {
 		&FlagPaymentsConsumerPricePerMinuteLowerBound,
 		&FlagPaymentsConsumerPricePerGBUpperBound,
 		&FlagPaymentsConsumerPricePerGBLowerBound,
+		&FlagPaymentsConsumerDataLeewayMegabytes,
 	)
 }
 
@@ -115,4 +122,5 @@ func ParseFlagsPayments(ctx *cli.Context) {
 	Current.ParseUInt64Flag(ctx, FlagPaymentsConsumerPricePerMinuteLowerBound)
 	Current.ParseUInt64Flag(ctx, FlagPaymentsConsumerPricePerGBUpperBound)
 	Current.ParseUInt64Flag(ctx, FlagPaymentsConsumerPricePerGBLowerBound)
+	Current.ParseUInt64Flag(ctx, FlagPaymentsConsumerDataLeewayMegabytes)
 }

core/node/options.go

@@ -138,6 +138,7 @@ func GetOptions() *Options {
 			ConsumerLowerGBPriceBound:          config.GetUInt64(config.FlagPaymentsConsumerPricePerGBLowerBound),
 			ConsumerUpperMinutePriceBound:      config.GetUInt64(config.FlagPaymentsConsumerPricePerMinuteUpperBound),
 			ConsumerLowerMinutePriceBound:      config.GetUInt64(config.FlagPaymentsConsumerPricePerMinuteLowerBound),
+			ConsumerDataLeewayMegabytes:        config.GetUInt64(config.FlagPaymentsConsumerDataLeewayMegabytes),
 		},
 		Accountant: OptionsAccountant{
 			AccountantID:              config.GetString(config.FlagAccountantID),

core/node/options_payments.go

@@ -31,4 +31,5 @@ type OptionsPayments struct {
 	ConsumerLowerGBPriceBound          uint64
 	ConsumerUpperMinutePriceBound      uint64
 	ConsumerLowerMinutePriceBound      uint64
+	ConsumerDataLeewayMegabytes        uint64
 }

services/openvpn/service/factory.go

@@ -34,7 +34,7 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-const statisticsReportingIntervalInSeconds = 30
+const statisticsReportingIntervalInSeconds = 1
 
 type eventBus interface {
 	Publish(topic string, data interface{})

services/wireguard/service/service_unix.go

@@ -208,7 +208,7 @@ func (m *Manager) ProvideConfig(sessionID string, sessionConfig json.RawMessage)
 		return nil, errors.Wrap(err, "failed to setup NAT/firewall rules")
 	}
 
-	statsPublisher := newStatsPublisher(m.publisher, 3*time.Second)
+	statsPublisher := newStatsPublisher(m.publisher, time.Second)
 	go statsPublisher.start(sessionID, conn)
 
 	destroy := func() {

session/pingpong/factory.go

@@ -24,6 +24,7 @@ import (
 	"github.com/mysteriumnetwork/node/communication"
 	"github.com/mysteriumnetwork/node/core/connection"
 	"github.com/mysteriumnetwork/node/core/node"
+	"github.com/mysteriumnetwork/node/datasize"
 	"github.com/mysteriumnetwork/node/eventbus"
 	"github.com/mysteriumnetwork/node/identity"
 	"github.com/mysteriumnetwork/node/market"
@@ -144,7 +145,8 @@ func ExchangeFactoryFunc(
 	channelImplementation string,
 	registryAddress string,
 	eventBus eventbus.EventBus,
-	getConsumerInfo getConsumerInfo) func(paymentInfo session.PaymentInfo,
+	getConsumerInfo getConsumerInfo,
+	dataLeewayMegabytes uint64) func(paymentInfo session.PaymentInfo,
 	dialog communication.Dialog,
 	consumer, provider, accountant identity.Identity, proposal market.ServiceProposal, sessionID string) (connection.PaymentIssuer, error) {
 	return func(paymentInfo session.PaymentInfo,
@@ -178,6 +180,7 @@ func ExchangeFactoryFunc(
 			AccountantAddress:         accountant,
 			ConsumerInfoGetter:        getConsumerInfo,
 			SessionID:                 sessionID,
+			DataLeeway:                datasize.MiB * datasize.BitSize(dataLeewayMegabytes),
 		}
 		return NewInvoicePayer(deps), nil
 	}

session/pingpong/invoice_payer.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/mysteriumnetwork/node/core/connection"
+	"github.com/mysteriumnetwork/node/datasize"
 	"github.com/mysteriumnetwork/node/eventbus"
 	"github.com/mysteriumnetwork/node/market"
 
@@ -45,6 +46,7 @@ var ErrProviderOvercharge = errors.New("provider is overcharging")
 
 const consumerFirstInvoiceTolerance = 1.35
 const consumerInvoiceTolerance = 1.05
+const dataLeeway = datasize.MiB * 20
 
 // PeerExchangeMessageSender allows for sending of exchange messages.
 type PeerExchangeMessageSender interface {
@@ -100,6 +102,7 @@ type InvoicePayerDeps struct {
 	EventBus                  eventbus.EventBus
 	AccountantAddress         identity.Identity
 	ConsumerInfoGetter        getConsumerInfo
+	DataLeeway                datasize.BitSize
 }
 
 // NewInvoicePayer returns a new instance of exchange message tracker.
@@ -226,7 +229,10 @@ func (ip *InvoicePayer) isInvoiceOK(invoice crypto.Invoice) error {
 		return ErrWrongProvider
 	}
 
-	shouldBe := calculatePaymentAmount(ip.deps.TimeTracker.Elapsed(), ip.getDataTransferred(), ip.deps.Proposal.PaymentMethod)
+	transfered := ip.getDataTransferred()
+	transfered.up += ip.deps.DataLeeway.Bytes()
+
+	shouldBe := calculatePaymentAmount(ip.deps.TimeTracker.Elapsed(), transfered, ip.deps.Proposal.PaymentMethod)
 
 	upperBound := uint64(math.Trunc(float64(shouldBe) * consumerInvoiceTolerance))
 	if !ip.receivedFirst {