call notifypayment in go routine in swap and restore full close (#953)

pkg/accounting/accounting.go

@@ -601,8 +601,7 @@ func surplusBalanceKeyPeer(key []byte) (swarm.Address, error) {
 	return addr, nil
 }
 
-// NotifyPayment implements the PaymentObserver interface. It is called by
-// Settlement when we receive a payment.
+// NotifyPayment is called by Settlement when we receive a payment.
 func (a *Accounting) NotifyPayment(peer swarm.Address, amount uint64) error {
 	accountingPeer, err := a.getAccountingPeer(peer)
 	if err != nil {
@@ -690,6 +689,18 @@ func (a *Accounting) NotifyPayment(peer swarm.Address, amount uint64) error {
 	return nil
 }
 
+// AsyncNotifyPayment calls notify payment in a go routine.
+// This is needed when accounting needs to be notified but the accounting lock is already held.
+func (a *Accounting) AsyncNotifyPayment(peer swarm.Address, amount uint64) error {
+	go func() {
+		err := a.NotifyPayment(peer, amount)
+		if err != nil {
+			a.logger.Errorf("failed to notify accounting of payment: %v", err)
+		}
+	}()
+	return nil
+}
+
 // subtractI64mU64 is a helper function for safe subtraction of Int64 - Uint64
 // It checks for
 //   - overflow safety in conversion of uint64 to int64

pkg/node/node.go

@@ -299,7 +299,7 @@ func NewBee(addr string, swarmAddress swarm.Address, publicKey ecdsa.PublicKey,
 		return nil, fmt.Errorf("accounting: %w", err)
 	}
 
-	settlement.SetPaymentObserver(acc)
+	settlement.SetNotifyPaymentFunc(acc.AsyncNotifyPayment)
 	pricing.SetPaymentThresholdObserver(acc)
 
 	kad := kademlia.New(swarmAddress, addressbook, hive, p2ps, logger, kademlia.Options{Bootnodes: bootnodes, Standalone: o.Standalone})

pkg/settlement/interface.go

@@ -28,12 +28,9 @@ type Interface interface {
 	SettlementsSent() (map[string]uint64, error)
 	// SettlementsReceived returns received settlements for each individual known peer
 	SettlementsReceived() (map[string]uint64, error)
-	// SetPaymentObserver sets the PaymentObserver to notify
-	SetPaymentObserver(observer PaymentObserver)
+	// SetNotifyPaymentFunc sets the NotifyPaymentFunc to notify
+	SetNotifyPaymentFunc(notifyPaymentFunc NotifyPaymentFunc)
 }
 
-// PaymentObserver is the interface Settlement uses to notify other components of an incoming payment
-type PaymentObserver interface {
-	// NotifyPayment is called when a payment from peer was successfully received
-	NotifyPayment(peer swarm.Address, amount uint64) error
-}
+// NotifyPaymentFunc is called when a payment from peer was successfully received
+type NotifyPaymentFunc func(peer swarm.Address, amount uint64) error

pkg/settlement/pseudosettle/mock/pseudosettle.go

@@ -115,7 +115,7 @@ func (s *Service) SettlementsReceived() (map[string]uint64, error) {
 	return s.settlementsRecv, nil
 }
 
-func (s *Service) SetPaymentObserver(settlement.PaymentObserver) {
+func (s *Service) SetNotifyPaymentFunc(settlement.NotifyPaymentFunc) {
 }
 
 // Option is the option passed to the mock settlement service

pkg/settlement/pseudosettle/pseudosettle.go

@@ -35,7 +35,7 @@ type Service struct {
 	streamer          p2p.Streamer
 	logger            logging.Logger
 	store             storage.StateStorer
-	observer settlement.PaymentObserver
+	notifyPaymentFunc settlement.NotifyPaymentFunc
 	metrics           metrics
 }
 
@@ -105,7 +105,7 @@ func (s *Service) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream) (e
 		return err
 	}
 
-	return s.observer.NotifyPayment(p.Address, req.Amount)
+	return s.notifyPaymentFunc(p.Address, req.Amount)
 }
 
 // Pay initiates a payment to the given peer
@@ -148,9 +148,9 @@ func (s *Service) Pay(ctx context.Context, peer swarm.Address, amount uint64) er
 	return nil
 }
 
-// SetPaymentObserver sets the payment observer which will be notified of incoming payments
-func (s *Service) SetPaymentObserver(observer settlement.PaymentObserver) {
-	s.observer = observer
+// SetNotifyPaymentFunc sets the NotifyPaymentFunc to notify
+func (s *Service) SetNotifyPaymentFunc(notifyPaymentFunc settlement.NotifyPaymentFunc) {
+	s.notifyPaymentFunc = notifyPaymentFunc
 }
 
 // TotalSent returns the total amount sent to a peer

pkg/settlement/pseudosettle/pseudosettle_test.go

@@ -47,7 +47,7 @@ func TestPayment(t *testing.T) {
 
 	observer := newTestObserver()
 	recipient := pseudosettle.New(nil, logger, storeRecipient)
-	recipient.SetPaymentObserver(observer)
+	recipient.SetNotifyPaymentFunc(observer.NotifyPayment)
 
 	recorder := streamtest.New(
 		streamtest.WithProtocols(recipient.Protocol()),

pkg/settlement/swap/mock/swap.go

@@ -24,7 +24,7 @@ type Service struct {
 
 	receiveChequeFunc    func(context.Context, swarm.Address, *chequebook.SignedCheque) error
 	payFunc              func(context.Context, swarm.Address, uint64) error
-	setPaymentObserverFunc func(observer settlement.PaymentObserver)
+	setNotifyPaymentFunc settlement.NotifyPaymentFunc
 	handshakeFunc        func(swarm.Address, common.Address) error
 	lastSentChequeFunc   func(swarm.Address) (*chequebook.SignedCheque, error)
 	lastSentChequesFunc  func() (map[string]*chequebook.SignedCheque, error)
@@ -75,9 +75,9 @@ func WithPayFunc(f func(context.Context, swarm.Address, uint64) error) Option {
 }
 
 // WithsettlementsFunc sets the mock settlements function
-func WithSetPaymentObserverFunc(f func(observer settlement.PaymentObserver)) Option {
+func WithSetNotifyPaymentFunc(f settlement.NotifyPaymentFunc) Option {
 	return optionFunc(func(s *Service) {
-		s.setPaymentObserverFunc = f
+		s.setNotifyPaymentFunc = f
 	})
 }
 
@@ -156,10 +156,9 @@ func (s *Service) Pay(ctx context.Context, peer swarm.Address, amount uint64) er
 	return nil
 }
 
-// SetPaymentObserver is the mock SetPaymentObserver function of swap.
-func (s *Service) SetPaymentObserver(observer settlement.PaymentObserver) {
-	if s.setPaymentObserverFunc != nil {
-		s.setPaymentObserverFunc(observer)
+func (s *Service) SetNotifyPaymentFunc(f settlement.NotifyPaymentFunc) {
+	if s.setNotifyPaymentFunc != nil {
+		s.SetNotifyPaymentFunc(f)
 	}
 }
 

pkg/settlement/swap/swap.go

@@ -50,7 +50,7 @@ type Service struct {
 	proto             swapprotocol.Interface
 	logger            logging.Logger
 	store             storage.StateStorer
-	observer    settlement.PaymentObserver
+	notifyPaymentFunc settlement.NotifyPaymentFunc
 	metrics           metrics
 	chequebook        chequebook.Service
 	chequeStore       chequebook.ChequeStore
@@ -102,7 +102,7 @@ func (s *Service) ReceiveCheque(ctx context.Context, peer swarm.Address, cheque
 
 	s.metrics.TotalReceived.Add(float64(amount.Uint64()))
 
-	return s.observer.NotifyPayment(peer, amount.Uint64())
+	return s.notifyPaymentFunc(peer, amount.Uint64())
 }
 
 // Pay initiates a payment to the given peer
@@ -129,9 +129,9 @@ func (s *Service) Pay(ctx context.Context, peer swarm.Address, amount uint64) er
 	return nil
 }
 
-// SetPaymentObserver sets the payment observer which will be notified of incoming payments
-func (s *Service) SetPaymentObserver(observer settlement.PaymentObserver) {
-	s.observer = observer
+// SetNotifyPaymentFunc sets the NotifyPaymentFunc to notify
+func (s *Service) SetNotifyPaymentFunc(notifyPaymentFunc settlement.NotifyPaymentFunc) {
+	s.notifyPaymentFunc = notifyPaymentFunc
 }
 
 // TotalSent returns the total amount sent to a peer

pkg/settlement/swap/swap_test.go

@@ -144,7 +144,7 @@ func TestReceiveCheque(t *testing.T) {
 	)
 
 	observer := &testObserver{}
-	swap.SetPaymentObserver(observer)
+	swap.SetNotifyPaymentFunc(observer.NotifyPayment)
 
 	err := swap.ReceiveCheque(context.Background(), peer, cheque)
 	if err != nil {
@@ -207,7 +207,7 @@ func TestReceiveChequeReject(t *testing.T) {
 	)
 
 	observer := &testObserver{}
-	swap.SetPaymentObserver(observer)
+	swap.SetNotifyPaymentFunc(observer.NotifyPayment)
 
 	err := swap.ReceiveCheque(context.Background(), peer, cheque)
 	if err == nil {
@@ -259,7 +259,7 @@ func TestReceiveChequeWrongChequebook(t *testing.T) {
 	)
 
 	observer := &testObserver{}
-	swapService.SetPaymentObserver(observer)
+	swapService.SetNotifyPaymentFunc(observer.NotifyPayment)
 
 	err := swapService.ReceiveCheque(context.Background(), peer, cheque)
 	if err == nil {

pkg/settlement/swap/swapprotocol/swapprotocol.go

@@ -186,8 +186,7 @@ func (s *Service) EmitCheque(ctx context.Context, peer swarm.Address, cheque *ch
 		if err != nil {
 			_ = stream.Reset()
 		} else {
-			// don't wait for full close to avoid deadlocks if cheques are sent simultaneously in both directions
-			go stream.FullClose()
+			_ = stream.FullClose()
 		}
 	}()