accounting: settle in reserve (#791)

pkg/accounting/accounting.go

@@ -28,12 +28,12 @@ var (
 
 // Interface is the Accounting interface.
 type Interface interface {
-	// Reserve reserves a portion of the balance for peer. Returns an error if
-	// the operation risks exceeding the disconnect threshold.
+	// Reserve reserves a portion of the balance for peer and attempts settlements if necessary.
+	// Returns an error if the operation risks exceeding the disconnect threshold or an attempted settlement failed.
 	//
-	// This should be called (always in combination with Release) before a
+	// This has to be called (always in combination with Release) before a
 	// Credit action to prevent overspending in case of concurrent requests.
-	Reserve(peer swarm.Address, price uint64) error
+	Reserve(ctx context.Context, peer swarm.Address, price uint64) error
 	// Release releases the reserved funds.
 	Release(peer swarm.Address, price uint64)
 	// Credit increases the balance the peer has with us (we "pay" the peer).
@@ -116,8 +116,8 @@ func NewAccounting(
 	}, nil
 }
 
-// Reserve reserves a portion of the balance for peer.
-func (a *Accounting) Reserve(peer swarm.Address, price uint64) error {
+// Reserve reserves a portion of the balance for peer and attempts settlements if necessary.
+func (a *Accounting) Reserve(ctx context.Context, peer swarm.Address, price uint64) error {
 	accountingPeer, err := a.getAccountingPeer(peer)
 	if err != nil {
 		return err
@@ -133,32 +133,54 @@ func (a *Accounting) Reserve(peer swarm.Address, price uint64) error {
 		}
 	}
 
+	// Check for safety of increase of reservedBalance by price
+	if accountingPeer.reservedBalance+price < accountingPeer.reservedBalance {
+		return ErrOverflow
+	}
+
+	nextReserved := accountingPeer.reservedBalance + price
+
 	// Subtract already reserved amount from actual balance, to get expected balance
-	expectedBalance, err := subtractI64mU64(currentBalance, accountingPeer.reservedBalance)
+	expectedBalance, err := subtractI64mU64(currentBalance, nextReserved)
 	if err != nil {
 		return err
 	}
 
-	// Determine if we owe anything to the peer, if we owe less than 0, we conclude we owe nothing
+	// Determine if we will owe anything to the peer, if we owe less than 0, we conclude we owe nothing
 	// This conversion is made safe by previous subtractI64mU64 not allowing MinInt64
 	expectedDebt := -expectedBalance
 	if expectedDebt < 0 {
 		expectedDebt = 0
 	}
 
-	// Check if the expected debt is already over the payment threshold.
-	if uint64(expectedDebt) > a.paymentThreshold {
-		a.metrics.AccountingBlocksCount.Inc()
-		return ErrOverdraft
+	threshold := accountingPeer.paymentThreshold
+	if threshold > a.earlyPayment {
+		threshold -= a.earlyPayment
+	} else {
+		threshold = 0
 	}
 
-	// Check for safety of increase of reservedBalance by price
-	if accountingPeer.reservedBalance+price < accountingPeer.reservedBalance {
-		return ErrOverflow
+	// If our expected debt is less than earlyPayment away from our payment threshold
+	// and we are actually in debt, trigger settlement.
+	// we pay early to avoid needlessly blocking request later when concurrent requests occur and we are already close to the payment threshold.
+	if expectedDebt >= int64(threshold) && currentBalance < 0 {
+		err = a.settle(ctx, peer, accountingPeer)
+		if err != nil {
+			return fmt.Errorf("failed to settle with peer %v: %v", peer, err)
+		}
+		// if we settled successfully our balance is back at 0
+		// and the expected debt therefore equals next reserved amount
+		expectedDebt = int64(nextReserved)
 	}
 
-	accountingPeer.reservedBalance += price
+	// if expectedDebt would still exceed the paymentThreshold at this point block this request
+	// this can happen if there is a large number of concurrent requests to the same peer
+	if expectedDebt > int64(a.paymentThreshold) {
+		a.metrics.AccountingBlocksCount.Inc()
+		return ErrOverdraft
+	}
 
+	accountingPeer.reservedBalance = nextReserved
 	return nil
 }
 
@@ -208,20 +230,6 @@ func (a *Accounting) Credit(peer swarm.Address, price uint64) error {
 
 	a.logger.Tracef("crediting peer %v with price %d, new balance is %d", peer, price, nextBalance)
 
-	// Get expectedbalance by safely decreasing current balance with reserved amounts
-	expectedBalance, err := subtractI64mU64(currentBalance, accountingPeer.reservedBalance)
-	if err != nil {
-		return err
-	}
-
-	// Compute expected debt before update because reserve still includes the
-	// amount that is deducted from the balance.
-	// This conversion is made safe by previous subtractI64mU64 not allowing MinInt64
-	expectedDebt := -expectedBalance
-	if expectedDebt < 0 {
-		expectedDebt = 0
-	}
-
 	err = a.store.Put(peerBalanceKey(peer), nextBalance)
 	if err != nil {
 		return fmt.Errorf("failed to persist balance: %w", err)
@@ -229,30 +237,12 @@ func (a *Accounting) Credit(peer swarm.Address, price uint64) error {
 
 	a.metrics.TotalCreditedAmount.Add(float64(price))
 	a.metrics.CreditEventsCount.Inc()
-
-	// If our expected debt is less than earlyPayment away from our payment threshold (which we assume is
-	// also the peers payment threshold), trigger settlement.
-	// we pay early to avoid needlessly blocking request later when concurrent requests occur and we are already close to the payment threshold
-	threshold := accountingPeer.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)
-		}
-	}
-
 	return nil
 }
 
 // Settle all debt with a peer. The lock on the accountingPeer must be held when
 // called.
-func (a *Accounting) settle(peer swarm.Address, balance *accountingPeer) error {
+func (a *Accounting) settle(ctx context.Context, peer swarm.Address, balance *accountingPeer) error {
 	oldBalance, err := a.Balance(peer)
 	if err != nil {
 		if !errors.Is(err, ErrPeerNoBalance) {
@@ -284,7 +274,7 @@ func (a *Accounting) settle(peer swarm.Address, balance *accountingPeer) error {
 		return fmt.Errorf("failed to persist balance: %w", err)
 	}
 
-	err = a.settlement.Pay(context.Background(), peer, paymentAmount)
+	err = a.settlement.Pay(ctx, peer, paymentAmount)
 	if err != nil {
 		err = fmt.Errorf("settlement for amount %d failed: %w", paymentAmount, err)
 		// If the payment didn't succeed we should restore the old balance in

pkg/accounting/accounting_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/ethersphere/bee/pkg/accounting"
 	"github.com/ethersphere/bee/pkg/logging"
 	"github.com/ethersphere/bee/pkg/p2p"
-	"github.com/ethersphere/bee/pkg/settlement"
+	mockSettlement "github.com/ethersphere/bee/pkg/settlement/pseudosettle/mock"
 	"github.com/ethersphere/bee/pkg/statestore/mock"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
@@ -65,7 +65,7 @@ func TestAccountingAddBalance(t *testing.T) {
 
 	for i, booking := range bookings {
 		if booking.price < 0 {
-			err = acc.Reserve(booking.peer, uint64(-booking.price))
+			err = acc.Reserve(context.Background(), booking.peer, uint64(-booking.price))
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -152,7 +152,7 @@ func TestAccountingAdd_persistentBalances(t *testing.T) {
 	}
 }
 
-// TestAccountingReserve tests that reserve returns an error if the payment threshold would be exceeded for a second time
+// TestAccountingReserve tests that reserve returns an error if the payment threshold would be exceeded
 func TestAccountingReserve(t *testing.T) {
 	logger := logging.New(ioutil.Discard, 0)
 
@@ -170,12 +170,7 @@ func TestAccountingReserve(t *testing.T) {
 	}
 
 	// it should allow to cross the threshold one time
-	err = acc.Reserve(peer1Addr, testPaymentThreshold+1)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = acc.Reserve(peer1Addr, 1)
+	err = acc.Reserve(context.Background(), peer1Addr, testPaymentThreshold+1)
 	if err == nil {
 		t.Fatal("expected error from reserve")
 	}
@@ -186,15 +181,15 @@ func TestAccountingReserve(t *testing.T) {
 }
 
 func TestAccountingOverflowReserve(t *testing.T) {
+
 	logger := logging.New(ioutil.Discard, 0)
 
 	store := mock.NewStateStore()
 	defer store.Close()
 
-	settlement := &settlementMock{}
+	settlement := mockSettlement.NewSettlement()
 
 	acc, err := accounting.NewAccounting(testPaymentThresholdLarge, 0, 0, logger, store, settlement, nil)
-
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -204,35 +199,30 @@ func TestAccountingOverflowReserve(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Try crediting near maximal value for peer
-	err = acc.Credit(peer1Addr, math.MaxInt64-2)
+	err = acc.Reserve(context.Background(), peer1Addr, testPaymentThresholdLarge)
 	if err != nil {
 		t.Fatal(err)
 	}
-	// Try reserving further maximal value
-	err = acc.Reserve(peer1Addr, math.MaxInt64)
+
+	err = acc.Reserve(context.Background(), peer1Addr, math.MaxInt64)
+	if !errors.Is(err, accounting.ErrOverflow) {
+		t.Fatalf("expected overflow error from Reserve, got %v", err)
+	}
+
+	acc.Release(peer1Addr, testPaymentThresholdLarge)
+
+	// Try crediting near maximal value for peer
+	err = acc.Credit(peer1Addr, math.MaxInt64)
 	if err != nil {
 		t.Fatal(err)
 	}
+
 	// Try reserving further value, should overflow
-	err = acc.Reserve(peer1Addr, 1)
-	if err == nil {
-		t.Fatal("expected error")
-	}
+	err = acc.Reserve(context.Background(), peer1Addr, 1)
 	// If we had other error, assert fail
 	if !errors.Is(err, accounting.ErrOverflow) {
-		t.Fatalf("expected overflow error from Debit, got %v", err)
+		t.Fatalf("expected overflow error from Reserve, got %v", err)
 	}
-	// Try reserving further near maximal value
-	err = acc.Reserve(peer1Addr, math.MaxInt64-2)
-	if err == nil {
-		t.Fatal("expected error")
-	}
-	// If we had other error, assert fail
-	if !errors.Is(err, accounting.ErrOverflow) {
-		t.Fatalf("expected overflow error from Debit, got %v", err)
-	}
-
 }
 
 func TestAccountingOverflowNotifyPayment(t *testing.T) {
@@ -241,7 +231,7 @@ func TestAccountingOverflowNotifyPayment(t *testing.T) {
 	store := mock.NewStateStore()
 	defer store.Close()
 
-	settlement := &settlementMock{}
+	settlement := mockSettlement.NewSettlement()
 
 	acc, err := accounting.NewAccounting(testPaymentThresholdLarge, 0, 0, logger, store, settlement, nil)
 	if err != nil {
@@ -389,36 +379,6 @@ func TestAccountingDisconnect(t *testing.T) {
 	}
 }
 
-type settlementMock struct {
-	paidAmount uint64
-	paidPeer   swarm.Address
-}
-
-func (s *settlementMock) Pay(ctx context.Context, peer swarm.Address, amount uint64) error {
-	s.paidPeer = peer
-	s.paidAmount = amount
-	return nil
-}
-
-func (s *settlementMock) TotalSent(peer swarm.Address) (totalSent uint64, err error) {
-	return 0, nil
-}
-
-func (s *settlementMock) TotalReceived(peer swarm.Address) (totalReceived uint64, err error) {
-	return 0, nil
-}
-
-func (s *settlementMock) SettlementsSent() (SettlementSent map[string]uint64, err error) {
-	return nil, nil
-}
-
-func (s *settlementMock) SettlementsReceived() (SettlementReceived map[string]uint64, err error) {
-	return nil, nil
-}
-
-func (s *settlementMock) SetPaymentObserver(settlement.PaymentObserver) {
-}
-
 // TestAccountingCallSettlement tests that settlement is called correctly if the payment threshold is hit
 func TestAccountingCallSettlement(t *testing.T) {
 	logger := logging.New(ioutil.Discard, 0)
@@ -426,7 +386,7 @@ func TestAccountingCallSettlement(t *testing.T) {
 	store := mock.NewStateStore()
 	defer store.Close()
 
-	settlement := &settlementMock{}
+	settlement := mockSettlement.NewSettlement()
 
 	acc, err := accounting.NewAccounting(testPaymentThreshold, 1000, 1000, logger, store, settlement, nil)
 	if err != nil {
@@ -438,7 +398,7 @@ func TestAccountingCallSettlement(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = acc.Reserve(peer1Addr, testPaymentThreshold)
+	err = acc.Reserve(context.Background(), peer1Addr, testPaymentThreshold)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -451,12 +411,21 @@ func TestAccountingCallSettlement(t *testing.T) {
 
 	acc.Release(peer1Addr, testPaymentThreshold)
 
-	if !settlement.paidPeer.Equal(peer1Addr) {
-		t.Fatalf("paid to wrong peer. got %v wanted %v", settlement.paidPeer, peer1Addr)
+	// try another request
+	err = acc.Reserve(context.Background(), peer1Addr, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	acc.Release(peer1Addr, 1)
+
+	totalSent, err := settlement.TotalSent(peer1Addr)
+	if err != nil {
+		t.Fatal(err)
 	}
 
-	if settlement.paidAmount != testPaymentThreshold {
-		t.Fatalf("paid wrong amount. got %d wanted %d", settlement.paidAmount, testPaymentThreshold)
+	if totalSent != testPaymentThreshold {
+		t.Fatalf("paid wrong amount. got %d wanted %d", totalSent, testPaymentThreshold)
 	}
 
 	balance, err := acc.Balance(peer1Addr)
@@ -468,14 +437,14 @@ func TestAccountingCallSettlement(t *testing.T) {
 	}
 
 	// Assume 100 is reserved by some other request
-	err = acc.Reserve(peer1Addr, 100)
+	err = acc.Reserve(context.Background(), peer1Addr, 100)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// Credit until the expected debt exceeeds payment threshold
 	expectedAmount := uint64(testPaymentThreshold - 100)
-	err = acc.Reserve(peer1Addr, expectedAmount)
+	err = acc.Reserve(context.Background(), peer1Addr, expectedAmount)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -485,13 +454,26 @@ func TestAccountingCallSettlement(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if !settlement.paidPeer.Equal(peer1Addr) {
-		t.Fatalf("paid to wrong peer. got %v wanted %v", settlement.paidPeer, peer1Addr)
+	acc.Release(peer1Addr, expectedAmount)
+
+	// try another request
+	err = acc.Reserve(context.Background(), peer1Addr, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	acc.Release(peer1Addr, 1)
+
+	totalSent, err = settlement.TotalSent(peer1Addr)
+	if err != nil {
+		t.Fatal(err)
 	}
 
-	if settlement.paidAmount != expectedAmount {
-		t.Fatalf("paid wrong amount. got %d wanted %d", settlement.paidAmount, expectedAmount)
+	if totalSent != expectedAmount+testPaymentThreshold {
+		t.Fatalf("paid wrong amount. got %d wanted %d", totalSent, expectedAmount+testPaymentThreshold)
 	}
+
+	acc.Release(peer1Addr, 100)
 }
 
 // TestAccountingCallSettlementEarly tests that settlement is called correctly if the payment threshold minus early payment is hit
@@ -501,7 +483,8 @@ func TestAccountingCallSettlementEarly(t *testing.T) {
 	store := mock.NewStateStore()
 	defer store.Close()
 
-	settlement := &settlementMock{}
+	settlement := mockSettlement.NewSettlement()
+	debt := uint64(500)
 	earlyPayment := uint64(1000)
 
 	acc, err := accounting.NewAccounting(testPaymentThreshold, testPaymentTolerance, earlyPayment, logger, store, settlement, nil)
@@ -514,26 +497,26 @@ func TestAccountingCallSettlementEarly(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	payment := testPaymentThreshold - earlyPayment
-	err = acc.Reserve(peer1Addr, payment)
+	err = acc.Credit(peer1Addr, debt)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Credit until payment treshold
-	err = acc.Credit(peer1Addr, payment)
+	payment := testPaymentThreshold - earlyPayment
+	err = acc.Reserve(context.Background(), 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)
+	totalSent, err := settlement.TotalSent(peer1Addr)
+	if err != nil {
+		t.Fatal(err)
 	}
 
-	if settlement.paidAmount != payment {
-		t.Fatalf("paid wrong amount. got %d wanted %d", settlement.paidAmount, payment)
+	if totalSent != debt {
+		t.Fatalf("paid wrong amount. got %d wanted %d", totalSent, testPaymentThreshold)
 	}
 
 	balance, err := acc.Balance(peer1Addr)
@@ -656,7 +639,7 @@ func TestAccountingNotifyPaymentThreshold(t *testing.T) {
 	defer store.Close()
 
 	pricing := &pricingMock{}
-	settlement := &settlementMock{}
+	settlement := mockSettlement.NewSettlement()
 
 	acc, err := accounting.NewAccounting(testPaymentThreshold, 1000, 0, logger, store, settlement, pricing)
 	if err != nil {
@@ -668,6 +651,7 @@ func TestAccountingNotifyPaymentThreshold(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	debt := uint64(50)
 	lowerThreshold := uint64(100)
 
 	err = acc.NotifyPaymentThreshold(peer1Addr, lowerThreshold)
@@ -675,17 +659,22 @@ func TestAccountingNotifyPaymentThreshold(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = acc.Reserve(peer1Addr, lowerThreshold)
+	err = acc.Credit(peer1Addr, debt)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = acc.Reserve(context.Background(), peer1Addr, lowerThreshold)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = acc.Credit(peer1Addr, lowerThreshold)
+	totalSent, err := settlement.TotalSent(peer1Addr)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	if settlement.paidAmount != lowerThreshold {
-		t.Fatalf("settled wrong amount. wanted %d, got %d", lowerThreshold, settlement.paidAmount)
+	if totalSent != debt {
+		t.Fatalf("paid wrong amount. got %d wanted %d", totalSent, debt)
 	}
 }

pkg/accounting/mock/accounting.go

@@ -5,6 +5,7 @@
 package mock
 
 import (
+	"context"
 	"sync"
 
 	"github.com/ethersphere/bee/pkg/accounting"
@@ -15,7 +16,7 @@ import (
 type Service struct {
 	lock         sync.Mutex
 	balances     map[string]int64
-	reserveFunc  func(peer swarm.Address, price uint64) error
+	reserveFunc  func(ctx context.Context, peer swarm.Address, price uint64) error
 	releaseFunc  func(peer swarm.Address, price uint64)
 	creditFunc   func(peer swarm.Address, price uint64) error
 	debitFunc    func(peer swarm.Address, price uint64) error
@@ -24,7 +25,7 @@ type Service struct {
 }
 
 // WithReserveFunc sets the mock Reserve function
-func WithReserveFunc(f func(peer swarm.Address, price uint64) error) Option {
+func WithReserveFunc(f func(ctx context.Context, peer swarm.Address, price uint64) error) Option {
 	return optionFunc(func(s *Service) {
 		s.reserveFunc = f
 	})
@@ -76,9 +77,9 @@ func NewAccounting(opts ...Option) accounting.Interface {
 }
 
 // Reserve is the mock function wrapper that calls the set implementation
-func (s *Service) Reserve(peer swarm.Address, price uint64) error {
+func (s *Service) Reserve(ctx context.Context, peer swarm.Address, price uint64) error {
 	if s.reserveFunc != nil {
-		return s.reserveFunc(peer, price)
+		return s.reserveFunc(ctx, peer, price)
 	}
 	return nil
 }

pkg/pushsync/pushsync.go

@@ -122,7 +122,7 @@ func (ps *PushSync) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream)
 
 	// compute the price we pay for this receipt and reserve it for the rest of this function
 	receiptPrice := ps.pricer.PeerPrice(peer, chunk.Address())
-	err = ps.accounting.Reserve(peer, receiptPrice)
+	err = ps.accounting.Reserve(ctx, peer, receiptPrice)
 	if err != nil {
 		return fmt.Errorf("reserve balance for peer %s: %w", peer.String(), err)
 	}
@@ -235,7 +235,7 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 
 	// compute the price we pay for this receipt and reserve it for the rest of this function
 	receiptPrice := ps.pricer.PeerPrice(peer, ch.Address())
-	err = ps.accounting.Reserve(peer, receiptPrice)
+	err = ps.accounting.Reserve(ctx, peer, receiptPrice)
 	if err != nil {
 		return nil, fmt.Errorf("reserve balance for peer %s: %w", peer.String(), err)
 	}

pkg/retrieval/retrieval.go

@@ -138,7 +138,7 @@ func (s *Service) retrieveChunk(ctx context.Context, addr swarm.Address, skipPee
 
 	// compute the price we pay for this chunk and reserve it for the rest of this function
 	chunkPrice := s.pricer.PeerPrice(peer, addr)
-	err = s.accounting.Reserve(peer, chunkPrice)
+	err = s.accounting.Reserve(ctx, peer, chunkPrice)
 	if err != nil {
 		return nil, peer, err
 	}

pkg/settlement/pseudosettle/mock/pseudosettle.go

@@ -23,6 +23,8 @@ type Service struct {
 
 	settlementsSentFunc func() (map[string]uint64, error)
 	settlementsRecvFunc func() (map[string]uint64, error)
+
+	payFunc func(context.Context, swarm.Address, uint64) error
 }
 
 // WithsettlementFunc sets the mock settlement function
@@ -51,6 +53,12 @@ func WithSettlementsRecvFunc(f func() (map[string]uint64, error)) Option {
 	})
 }
 
+func WithPayFunc(f func(context.Context, swarm.Address, uint64) error) Option {
+	return optionFunc(func(s *Service) {
+		s.payFunc = f
+	})
+}
+
 // Newsettlement creates the mock settlement implementation
 func NewSettlement(opts ...Option) settlement.Interface {
 	mock := new(Service)
@@ -62,7 +70,10 @@ func NewSettlement(opts ...Option) settlement.Interface {
 	return mock
 }
 
-func (s *Service) Pay(_ context.Context, peer swarm.Address, amount uint64) error {
+func (s *Service) Pay(c context.Context, peer swarm.Address, amount uint64) error {
+	if s.payFunc != nil {
+		return s.payFunc(c, peer, amount)
+	}
 	s.settlementsSent[peer.String()] += amount
 	return nil
 }