diff --git a/cmd/di.go b/cmd/di.go index 03d9b46e29..3ec0cae099 100644 --- a/cmd/di.go +++ b/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, diff --git a/config/flags_payments.go b/config/flags_payments.go index 3cc55a04b0..6c456b974a 100644 --- a/config/flags_payments.go +++ b/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) } diff --git a/core/node/options.go b/core/node/options.go index 76d2a3b0b7..13a9a867e0 100644 --- a/core/node/options.go +++ b/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), diff --git a/core/node/options_payments.go b/core/node/options_payments.go index 3bdbbe68b2..d4d221aa5e 100644 --- a/core/node/options_payments.go +++ b/core/node/options_payments.go @@ -31,4 +31,5 @@ type OptionsPayments struct { ConsumerLowerGBPriceBound uint64 ConsumerUpperMinutePriceBound uint64 ConsumerLowerMinutePriceBound uint64 + ConsumerDataLeewayMegabytes uint64 } diff --git a/services/openvpn/service/factory.go b/services/openvpn/service/factory.go index cb4023cedf..4c10ac67f5 100644 --- a/services/openvpn/service/factory.go +++ b/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{}) diff --git a/services/wireguard/service/service_unix.go b/services/wireguard/service/service_unix.go index 9975711caa..987a8358be 100644 --- a/services/wireguard/service/service_unix.go +++ b/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() { diff --git a/session/pingpong/factory.go b/session/pingpong/factory.go index 2d6b96a540..b015eba6d3 100644 --- a/session/pingpong/factory.go +++ b/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 } diff --git a/session/pingpong/invoice_payer.go b/session/pingpong/invoice_payer.go index 3acea8e78c..29f03917a7 100644 --- a/session/pingpong/invoice_payer.go +++ b/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 {