Merge branch 'develop' into release/v0.26.0

blockchain/abci/abci.go

@@ -54,16 +54,9 @@ func (app *App) Commit() (resp types.ResponseCommit) {
 }
 
 func (app *App) CheckTx(req types.RequestCheckTx) (resp types.ResponseCheckTx) {
-	tx := app.txFromCache(req.Tx)
-	if tx == nil {
-		var (
-			code uint32
-			err  error
-		)
-		tx, code, err = app.decodeAndValidateTx(req.GetTx())
-		if err != nil {
-			return NewResponseCheckTx(code, err.Error())
-		}
+	tx, code, err := app.getTx(req.GetTx())
+	if err != nil {
+		return NewResponseCheckTx(code, err.Error())
 	}
 
 	ctx := app.ctx
@@ -90,23 +83,15 @@ func (app *App) CheckTx(req types.RequestCheckTx) (resp types.ResponseCheckTx) {
 }
 
 func (app *App) DeliverTx(req types.RequestDeliverTx) (resp types.ResponseDeliverTx) {
-	tx := app.txFromCache(req.Tx)
-	if tx == nil {
-		var (
-			code uint32
-			err  error
-		)
-		tx, code, err = app.decodeAndValidateTx(req.GetTx())
-		if err != nil {
-			return NewResponseDeliverTx(code, err.Error())
-		}
-	} else {
-		app.removeTxFromCache(req.Tx)
+	tx, code, err := app.getTx(req.GetTx())
+	if err != nil {
+		return NewResponseDeliverTx(code, err.Error())
 	}
 
 	if err := app.replayProtector.DeliverTx(tx); err != nil {
 		return NewResponseDeliverTx(AbciTxnValidationFailure, err.Error())
 	}
+	app.removeTxFromCache(req.GetTx())
 
 	// It's been validated by CheckTx so we can skip the validation here
 	ctx := app.ctx

blockchain/abci/app.go

@@ -101,3 +101,22 @@ func (app *App) removeTxFromCache(in []byte) {
 	key := string(in)
 	delete(app.checkedTxs, key)
 }
+
+// getTx returns an internal Tx given a []byte.
+// if no errors were found during decoding and validation, the resulting Tx
+// will be cached.
+// An error code different from 0 is returned if decoding or validation fails
+// with its the corresponding error
+func (app *App) getTx(bytes []byte) (Tx, uint32, error) {
+	if tx := app.txFromCache(bytes); tx != nil {
+		return tx, 0, nil
+	}
+
+	tx, code, err := app.decodeAndValidateTx(bytes)
+	if err != nil {
+		return nil, code, err
+	}
+
+	app.cacheTx(bytes, tx)
+	return tx, 0, nil
+}

storage/trades.go

@@ -381,6 +381,10 @@ func (ts *Trade) notify(items []types.Trade) error {
 	if len(items) == 0 {
 		return nil
 	}
+
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
+
 	if len(ts.subscribers) == 0 {
 		ts.log.Debug("No subscribers connected in trade store")
 		return nil