Fix nil exception (#718)

* Fix nil exception when accessing cached tx status. Introduced service that fetches the tx status to distinguish various cases where different inputs are needed.

Signed-off-by: Alexandros Filios <[email protected]>

Author: alexandrosfilios

token/services/db/retry.go

@@ -9,27 +9,33 @@ package db
 import (
 	"errors"
 	"time"
+
+	logging2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/logging"
 )
 
 // RetryRunner receives a function that potentially fails and retries according to the specified strategy
 type RetryRunner interface {
 	Run(func() error) error
 }
 
+var ErrMaxRetriesExceeded = errors.New("maximum number of retries exceeded")
+
 const Infinitely = -1
 
 func NewRetryRunner(maxTimes int, delay time.Duration, expBackoff bool) *retryRunner {
 	return &retryRunner{
 		delay:      delay,
 		expBackoff: expBackoff,
 		maxTimes:   maxTimes,
+		logger:     logging2.MustGetLogger("retry-runner"),
 	}
 }
 
 type retryRunner struct {
 	delay      time.Duration
 	expBackoff bool
 	maxTimes   int
+	logger     logging2.Logger
 }
 
 func (f *retryRunner) nextDelay() time.Duration {
@@ -40,14 +46,30 @@ func (f *retryRunner) nextDelay() time.Duration {
 }
 
 func (f *retryRunner) Run(runner func() error) error {
+	return f.RunWithErrors(func() (bool, error) {
+		err := runner()
+		return err == nil, err
+	})
+}
+
+// RunWithErrors will retry until runner() returns true or until it returns maxTimes false.
+// If it returns true, then the error or nil will be returned.
+// If it returns maxTimes false, then it will always return an error: either a join of all errors it encountered or a ErrMaxRetriesExceeded.
+func (f *retryRunner) RunWithErrors(runner func() (bool, error)) error {
 	errs := make([]error, 0)
 	for i := 0; f.maxTimes < 0 || i < f.maxTimes; i++ {
-		if err := runner(); err != nil {
+		terminate, err := runner()
+		if terminate {
+			return err
+		}
+		if err != nil {
 			errs = append(errs, err)
-			time.Sleep(f.nextDelay())
-		} else {
-			return nil
 		}
+		f.logger.Debugf("Will retry iteration [%d] after delay. %d errors returned so far", i+1, len(errs))
+		time.Sleep(f.nextDelay())
+	}
+	if len(errs) == 0 {
+		return ErrMaxRetriesExceeded
 	}
 	return errors.Join(errs...)
 }

token/services/network/orion/approval.go

@@ -16,6 +16,7 @@ import (
 	"github.com/hyperledger-labs/fabric-token-sdk/token"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/core/common"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/driver"
+	"github.com/hyperledger-labs/fabric-token-sdk/token/services/db"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/services/network"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/services/network/common/rws/translator"
 	driver2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/network/driver"
@@ -24,6 +25,7 @@ import (
 
 type TxStatusResponseCache interface {
 	Get(key string) (*TxStatusResponse, bool)
+	GetOrLoad(key string, loader func() (*TxStatusResponse, error)) (*TxStatusResponse, bool, error)
 	Add(key string, value *TxStatusResponse)
 }
 
@@ -153,46 +155,49 @@ func (r *RequestApprovalResponderView) process(context view.Context, request *Ap
 
 	// commit
 	txStatusFetcher := &RequestTxStatusResponderView{dbManager: r.dbManager, statusCache: r.statusCache}
-	numRetries := 5
-	sleepDuration := 1 * time.Second
-	for i := 0; i < numRetries; i++ {
+
+	runner := db.NewRetryRunner(5, time.Second, true)
+
+	var envelopeRaw []byte
+	validateErr := runner.RunWithErrors(func() (bool, error) {
 		span.AddEvent("try_validate")
-		envelopeRaw, retry, err := r.validate(context, request, validator)
+		var retry bool
+		envelopeRaw, retry, err = r.validate(context, request, validator)
+		if err == nil {
+			return true, nil
+		}
+
+		if !retry {
+			logger.Errorf("failed to commit transaction [%s], no more retry after this", err)
+			return true, errors.Wrapf(err, "failed to commit transaction [%s]", request.TxID)
+		}
+		logger.Errorf("failed to commit transaction [%s], retry", err)
+		// was the transaction committed, by any chance?
+		span.AddEvent("fetch_tx_status")
+		status, err := txStatusFetcher.process(context, &TxStatusRequest{
+			Network:   request.Network,
+			Namespace: request.Namespace,
+			TxID:      request.TxID,
+		})
 		if err != nil {
-			if !retry {
-				logger.Errorf("failed to commit transaction [%s], no more retry after this [%d]", err, i)
-				return nil, errors.Wrapf(err, "failed to commit transaction [%s]", request.TxID)
-			}
-			logger.Errorf("failed to commit transaction [%s], retry [%d]", err, i)
-			// was the transaction committed, by any chance?
-			span.AddEvent("fetch_tx_status")
-			status, err := txStatusFetcher.process(context, &TxStatusRequest{
-				Network:   request.Network,
-				Namespace: request.Namespace,
-				TxID:      request.TxID,
-			})
-			if err != nil {
-				logger.Errorf("failed to ask transaction status [%s], retry [%d]", err, i)
-			}
-			if status != nil {
-				if status.Status == network.Valid {
-					return nil, nil
-				}
-				if status.Status == network.Invalid {
-					break
-				}
-				logger.Debugf("transaction [%s] status [%s], retry [%d], wait a bit and resubmit", request.TxID, status, i)
-			} else {
-				logger.Errorf("failed to ask transaction status [%s], got a nil answert, retry [%d]", request.TxID, i)
-			}
-			time.Sleep(sleepDuration)
-			sleepDuration = sleepDuration * 2
-			continue
+			logger.Errorf("failed to ask transaction status [%s], retry", err)
+			return false, nil
 		}
-		return envelopeRaw, nil
-	}
 
-	return nil, errors.Wrapf(err, "failed to commit transaction [%s]", request.TxID)
+		if status.Status == network.Valid {
+			return true, nil
+		}
+		if status.Status == network.Invalid {
+			return true, errors.New("invalid transaction status")
+		}
+		logger.Debugf("transaction [%s] status [%s], retry, wait a bit and resubmit", request.TxID, status)
+		return false, nil
+	})
+
+	if validateErr != nil {
+		return nil, errors.Wrapf(err, "failed to commit transaction [%s]", request.TxID)
+	}
+	return envelopeRaw, nil
 }
 
 func (r *RequestApprovalResponderView) validate(context view.Context, request *ApprovalRequest, validator driver.Validator) ([]byte, bool, error) {

token/services/network/orion/broadcast.go

@@ -14,6 +14,7 @@ import (
 	view2 "github.com/hyperledger-labs/fabric-smart-client/platform/view"
 	session2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/session"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
+	"github.com/hyperledger-labs/fabric-token-sdk/token/services/db"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/services/network"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/services/network/driver"
 	"github.com/pkg/errors"
@@ -107,73 +108,66 @@ func (r *BroadcastResponderView) Call(context view.Context) (interface{}, error)
 		return nil, errors.Wrapf(err, "failed to get session manager for [%s]", request.Network)
 	}
 
-	done := false
-	err = nil
-	txStatusFetcher := &RequestTxStatusResponderView{dbManager: r.dbManager, statusCache: r.statusCache}
-	numRetries := 5
-	sleepDuration := 1 * time.Second
+	txStatusFetcher := NewStatusFetcher(r.dbManager)
+
+	runner := db.NewRetryRunner(5, 1*time.Second, true)
+
 	var txID string
-	for i := 0; i < numRetries; i++ {
+	broadcastErr := runner.RunWithErrors(func() (bool, error) {
 		span.AddEvent("try_broadcast")
-		var err2 error
-		if _, txID, err2 = r.broadcast(context, sm, request); err2 != nil {
-			span.RecordError(err2)
-			logger.Errorf("failed to broadcast to [%s], txID [%s] with err [%s], retry [%d]", sm.CustodianID, txID, err2, i)
-			if strings.Contains(err2.Error(), "is not valid") {
-				err = err2
-				break
-			}
-			if len(txID) != 0 {
-				// was the transaction committed, by any chance?
-				logger.Errorf("check transaction [%s] status on [%s], retry [%d]", txID, sm.CustodianID, i)
-				span.AddEvent("fetch_tx_status")
-				status, err := txStatusFetcher.process(context, &TxStatusRequest{
-					Network: request.Network,
-					TxID:    txID,
-				})
-				if err != nil {
-					logger.Errorf("failed to ask transaction status [%s][%s], retry [%d]", txID, err, i)
-				}
-				if status != nil {
-					if status.Status == network.Valid {
-						done = true
-						break
-					}
-					if status.Status == network.Invalid {
-						break
-					}
-					logger.Debugf("transaction [%s] status [%d], retry [%d], wait a bit and resubmit", txID, status, i)
-				} else {
-					logger.Errorf("failed to ask transaction status [%s], got a nil answert, retry [%d]", txID, i)
-				}
-			}
-			time.Sleep(sleepDuration)
-			sleepDuration = sleepDuration * 2
-			continue
+		var err error
+		_, txID, err = r.broadcast(context, sm, request)
+		if err == nil {
+			return true, nil
 		}
-		done = true
-		break
-	}
-	var broadcastError string
-	if !done {
-		broadcastError = fmt.Sprintf("failed to broadcast to [%s] with err [%s]", sm.CustodianID, err)
-	}
+
+		span.RecordError(err)
+		logger.Errorf("failed to broadcast to [%s], txID [%s] with err [%s], retry", sm.CustodianID, txID, err)
+		if strings.Contains(err.Error(), "is not valid") {
+			return true, err
+		}
+		if len(txID) == 0 {
+			return false, nil
+		}
+
+		// was the transaction committed, by any chance?
+		logger.Errorf("check transaction [%s] status on [%s], retry", txID, sm.CustodianID)
+		span.AddEvent("fetch_tx_status")
+
+		status, err := txStatusFetcher.FetchCode(request.Network, txID)
+		if err != nil {
+			logger.Errorf("failed to ask transaction status [%s][%s]", txID, err)
+			return false, nil
+		}
+		if status == network.Valid {
+			return true, nil
+		}
+		if status == network.Invalid {
+			return true, errors.New("invalid transaction status")
+		}
+		logger.Debugf("transaction [%s] status [%d], retry, wait a bit and resubmit", txID, status)
+		return false, nil
+	})
 
 	// update cache
 	if len(txID) != 0 {
 		response, ok := r.statusCache.Get(txID)
 		if ok {
-			if len(broadcastError) == 0 {
+			if broadcastErr == nil {
 				response.Status = driver.Valid
 			} else {
 				response.Status = driver.Invalid
 			}
+			r.statusCache.Add(txID, response)
 		}
-		r.statusCache.Add(txID, response)
 	}
 
 	// send back answer
-	if err := session.SendWithContext(context.Context(), &BroadcastResponse{Err: broadcastError}); err != nil {
+	broadcastResponse := &BroadcastResponse{}
+	if broadcastErr != nil {
+		broadcastResponse.Err = fmt.Sprintf("failed to broadcast to [%s] with err [%s]", sm.CustodianID, broadcastErr.Error())
+	}
+	if err := session.SendWithContext(context.Context(), broadcastResponse); err != nil {
 		return nil, errors.Wrapf(err, "failed to send response")
 	}
 	return nil, nil

token/services/network/orion/statusfetcher.go

@@ -0,0 +1,87 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package orion
+
+import (
+	"encoding/base64"
+
+	errors2 "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
+	"github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
+	"github.com/hyperledger-labs/fabric-smart-client/platform/orion"
+	driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/orion/driver"
+	"github.com/hyperledger-labs/fabric-token-sdk/token/services/network/common/rws/keys"
+	"github.com/hyperledger-labs/orion-sdk-go/pkg/bcdb"
+	"github.com/pkg/errors"
+	"go.uber.org/zap/zapcore"
+)
+
+type StatusFetcher struct {
+	dbManager *DBManager
+}
+
+func NewStatusFetcher(dbManager *DBManager) *StatusFetcher {
+	return &StatusFetcher{dbManager: dbManager}
+}
+
+func (r *StatusFetcher) FetchStatus(network, namespace string, txID driver.TxID) (*TxStatusResponse, error) {
+	oSession, code, err := r.fetchCode(network, txID)
+	if err != nil {
+		return nil, err
+	}
+
+	// fetch token request reference
+	qe, err := oSession.QueryExecutor(namespace)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get query executor [%s] for orion network [%s]", txID, network)
+	}
+	key, err := keys.CreateTokenRequestKey(txID)
+	if err != nil {
+		return nil, errors.Errorf("can't create for token request '%s'", txID)
+	}
+	trRef, err := qe.Get(orionKey(key))
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get token request reference [%s] for orion network [%s]", txID, network)
+	}
+	if logger.IsEnabledFor(zapcore.DebugLevel) {
+		logger.Debugf("retrieved token request hash for [%s][%s]:[%s]", key, txID, base64.StdEncoding.EncodeToString(trRef))
+	}
+	return &TxStatusResponse{
+		TokenRequestReference: trRef,
+		Status:                code,
+	}, nil
+}
+
+func (r *StatusFetcher) FetchCode(network string, txID driver.TxID) (driver2.ValidationCode, error) {
+	_, code, err := r.fetchCode(network, txID)
+	return code, err
+}
+
+func (r *StatusFetcher) fetchCode(network string, txID driver.TxID) (*orion.Session, driver2.ValidationCode, error) {
+	sm, err := r.dbManager.GetSessionManager(network)
+	if err != nil {
+		return nil, 0, errors.Wrapf(err, "failed to get session manager for network [%s]", network)
+	}
+	oSession, err := sm.GetSession()
+	if err != nil {
+		return nil, 0, errors.Wrapf(err, "failed to create session to orion network [%s]", network)
+	}
+	ledger, err := oSession.Ledger()
+	if err != nil {
+		return nil, 0, errors.Wrapf(err, "failed to get ledger for orion network [%s]", network)
+	}
+	tx, err := ledger.GetTransactionByID(txID)
+	if err != nil {
+		if errors2.HasType(err, &bcdb.ErrorNotFound{}) {
+			return oSession, driver2.Unknown, nil
+		}
+		return nil, 0, errors.Wrapf(err, "failed to get transaction [%s] for orion network [%s]", txID, network)
+	}
+	if tx.ValidationCode() == orion.VALID {
+		return oSession, driver2.Valid, nil
+	}
+	return oSession, driver2.Invalid, nil
+}