trigger settlement before expected debt hits payment threshold (#745)

cmd/bee/cmd/cmd.go

@@ -40,6 +40,7 @@ const (
 	optionNameGlobalPinningEnabled = "global-pinning-enable"
 	optionNamePaymentThreshold     = "payment-threshold"
 	optionNamePaymentTolerance     = "payment-tolerance"
+	optionNamePaymentEarly         = "payment-early"
 	optionNameResolverEndpoints    = "resolver-options"
 	optionNameGatewayMode          = "gateway-mode"
 	optionNameClefSignerEnable     = "clef-signer-enable"
@@ -190,6 +191,7 @@ func (c *command) setAllFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool(optionNameGlobalPinningEnabled, false, "enable global pinning")
 	cmd.Flags().Uint64(optionNamePaymentThreshold, 100000, "threshold in BZZ where you expect to get paid from your peers")
 	cmd.Flags().Uint64(optionNamePaymentTolerance, 10000, "excess debt above payment threshold in BZZ where you disconnect from your peer")
+	cmd.Flags().Uint64(optionNamePaymentEarly, 10000, "amount in BZZ below the peers payment threshold when we initiate settlement")
 	cmd.Flags().StringSlice(optionNameResolverEndpoints, []string{}, "resolver connection string, see help for format")
 	cmd.Flags().Bool(optionNameGatewayMode, false, "disable a set of sensitive features in the api")
 	cmd.Flags().Bool(optionNameClefSignerEnable, false, "enable clef signer")

pkg/accounting/accounting.go

@@ -57,6 +57,7 @@ type accountingPeer struct {
 type Options struct {
 	PaymentThreshold uint64
 	PaymentTolerance uint64
+	EarlyPayment     uint64
 	Logger           logging.Logger
 	Store            storage.StateStorer
 	Settlement       settlement.Interface
@@ -74,6 +75,7 @@ type Accounting struct {
 	// The amount in BZZ we let peers exceed the payment threshold before we
 	// disconnect them.
 	paymentTolerance uint64
+	earlyPayment     uint64
 	settlement       settlement.Interface
 	metrics          metrics
 }
@@ -106,6 +108,7 @@ func NewAccounting(o Options) (*Accounting, error) {
 		accountingPeers:  make(map[string]*accountingPeer),
 		paymentThreshold: o.PaymentThreshold,
 		paymentTolerance: o.PaymentTolerance,
+		earlyPayment:     o.EarlyPayment,
 		logger:           o.Logger,
 		store:            o.Store,
 		settlement:       o.Settlement,
@@ -227,9 +230,17 @@ func (a *Accounting) Credit(peer swarm.Address, price uint64) error {
 	a.metrics.TotalCreditedAmount.Add(float64(price))
 	a.metrics.CreditEventsCount.Inc()
 
-	// If our expected debt exceeds our payment threshold (which we assume is
+	// If our expected debt is less than earlyPayment away from our payment threshold (which we assume is
 	// also the peers payment threshold), trigger settlement.
-	if uint64(expectedDebt) >= a.paymentThreshold {
+	// we pay early to avoid needlessly blocking request later when concurrent requests occur and we are already close to the payment threshold
+	threshold := a.paymentThreshold
+	if threshold > a.earlyPayment {
+		threshold -= a.earlyPayment
+	} else {
+		threshold = 0
+	}
+
+	if uint64(expectedDebt) >= threshold {
 		err = a.settle(peer, accountingPeer)
 		if err != nil {
 			a.logger.Errorf("failed to settle with peer %v: %v", peer, err)

pkg/accounting/accounting_test.go

@@ -536,6 +536,64 @@ func TestAccountingCallSettlement(t *testing.T) {
 	}
 }
 
+// TestAccountingCallSettlementEarly tests that settlement is called correctly if the payment threshold minus early payment is hit
+func TestAccountingCallSettlementEarly(t *testing.T) {
+	logger := logging.New(ioutil.Discard, 0)
+
+	store := mock.NewStateStore()
+	defer store.Close()
+
+	settlement := &settlementMock{}
+	earlyPayment := uint64(1000)
+
+	acc, err := accounting.NewAccounting(accounting.Options{
+		PaymentThreshold: testPaymentThreshold,
+		PaymentTolerance: testPaymentTolerance,
+		EarlyPayment:     earlyPayment,
+		Logger:           logger,
+		Store:            store,
+		Settlement:       settlement,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	peer1Addr, err := swarm.ParseHexAddress("00112233")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	payment := testPaymentThreshold - earlyPayment
+	err = acc.Reserve(peer1Addr, payment)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Credit until payment treshold
+	err = acc.Credit(peer1Addr, payment)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	acc.Release(peer1Addr, payment)
+
+	if !settlement.paidPeer.Equal(peer1Addr) {
+		t.Fatalf("paid to wrong peer. got %v wanted %v", settlement.paidPeer, peer1Addr)
+	}
+
+	if settlement.paidAmount != payment {
+		t.Fatalf("paid wrong amount. got %d wanted %d", settlement.paidAmount, payment)
+	}
+
+	balance, err := acc.Balance(peer1Addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if balance != 0 {
+		t.Fatalf("expected balance to be reset. got %d", balance)
+	}
+}
+
 // TestAccountingNotifyPayment tests that payments adjust the balance and payment which put us into debt are rejected
 func TestAccountingNotifyPayment(t *testing.T) {
 	logger := logging.New(ioutil.Discard, 0)

pkg/node/node.go

@@ -96,6 +96,7 @@ type Options struct {
 	GlobalPinningEnabled   bool
 	PaymentThreshold       uint64
 	PaymentTolerance       uint64
+	PaymentEarly           uint64
 	ResolverConnectionCfgs []multiresolver.ConnectionConfig
 	GatewayMode            bool
 	SwapEndpoint           string
@@ -296,6 +297,7 @@ func NewBee(addr string, swarmAddress swarm.Address, keystore keystore.Service,
 		Store:            stateStore,
 		PaymentThreshold: o.PaymentThreshold,
 		PaymentTolerance: o.PaymentTolerance,
+		EarlyPayment:     o.PaymentEarly,
 		Settlement:       settlement,
 	})
 	if err != nil {