this commit removes as many panic as possible (#441)

Signed-off-by: Angelo De Caro <adc@zurich.ibm.com>

Author: adecaro

platform/fabric/core/generic/channel.go

@@ -125,10 +125,9 @@ func newChannel(network *network, name string, quiet bool) (*channel, error) {
 
 	// Delivery
 	deliveryService, err := delivery2.New(name, sp, network, func(block *common.Block) (bool, error) {
-		if err := committerInst.Commit(block); err != nil {
-			return true, err
-		}
-		return false, nil
+		// commit the block, if an error occurs then retry
+		err := committerInst.Commit(block)
+		return false, err
 	}, txIDStore, waitForEventTimeout)
 	if err != nil {
 		return nil, err

platform/fabric/core/generic/committer/committer.go

@@ -8,7 +8,6 @@ package committer
 
 import (
 	"context"
-	"fmt"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -27,7 +26,7 @@ const (
 	ConfigTXPrefix = "configtx_"
 )
 
-var logger = flogging.MustGetLogger("fabric-sdk.committer")
+var logger = flogging.MustGetLogger("fabric-sdk.Committer")
 
 type Metrics interface {
 	EmitKey(val float32, event ...string)
@@ -43,7 +42,7 @@ type Network interface {
 	Ledger(channel string) (driver.Ledger, error)
 }
 
-type committer struct {
+type Committer struct {
 	channel             string
 	network             Network
 	finality            Finality
@@ -58,12 +57,12 @@ type committer struct {
 	publisher      events.Publisher
 }
 
-func New(channel string, network Network, finality Finality, waitForEventTimeout time.Duration, quiet bool, metrics Metrics, publisher events.Publisher) (*committer, error) {
+func New(channel string, network Network, finality Finality, waitForEventTimeout time.Duration, quiet bool, metrics Metrics, publisher events.Publisher) (*Committer, error) {
 	if len(channel) == 0 {
-		panic("expected a channel, got empty string")
+		return nil, errors.Errorf("expected a channel, got empty string")
 	}
 
-	d := &committer{
+	d := &Committer{
 		channel:             channel,
 		network:             network,
 		waitForEventTimeout: waitForEventTimeout,
@@ -79,7 +78,7 @@ func New(channel string, network Network, finality Finality, waitForEventTimeout
 }
 
 // Commit commits the transactions in the block passed as argument
-func (c *committer) Commit(block *common.Block) error {
+func (c *Committer) Commit(block *common.Block) error {
 	for i, tx := range block.Data.Data {
 
 		env, err := protoutil.UnmarshalEnvelope(tx)
@@ -100,27 +99,31 @@ func (c *committer) Commit(block *common.Block) error {
 
 		var event TxEvent
 
-		c.metrics.EmitKey(0, "committer", "start", "Commit", chdr.TxId)
+		c.metrics.EmitKey(0, "Committer", "start", "Commit", chdr.TxId)
 		switch common.HeaderType(chdr.Type) {
 		case common.HeaderType_CONFIG:
 			if logger.IsEnabledFor(zapcore.DebugLevel) {
 				logger.Debugf("[%s] Config transaction received: %s", c.channel, chdr.TxId)
 			}
-			c.handleConfig(block, i, env)
+			if err := c.handleConfig(block, i, env); err != nil {
+				return err
+			}
 		case common.HeaderType_ENDORSER_TRANSACTION:
 			if logger.IsEnabledFor(zapcore.DebugLevel) {
 				logger.Debugf("[%s] Endorser transaction received: %s", c.channel, chdr.TxId)
 			}
 			if len(block.Metadata.Metadata) < int(common.BlockMetadataIndex_TRANSACTIONS_FILTER) {
 				return errors.Errorf("block metadata lacks transaction filter")
 			}
-			c.handleEndorserTransaction(block, i, &event, env, chdr)
+			if err := c.handleEndorserTransaction(block, i, &event, env, chdr); err != nil {
+				return err
+			}
 		default:
 			if logger.IsEnabledFor(zapcore.DebugLevel) {
 				logger.Debugf("[%s] Received unhandled transaction type: %s", c.channel, chdr.Type)
 			}
 		}
-		c.metrics.EmitKey(0, "committer", "end", "Commit", chdr.TxId)
+		c.metrics.EmitKey(0, "Committer", "end", "Commit", chdr.TxId)
 
 		c.notify(event)
 
@@ -135,9 +138,9 @@ func (c *committer) Commit(block *common.Block) error {
 // IsFinal takes in input a transaction id and waits for its confirmation
 // with the respect to the passed context that can be used to set a deadline
 // for the waiting time.
-func (c *committer) IsFinal(ctx context.Context, txID string) error {
-	c.metrics.EmitKey(0, "committer", "start", "IsFinal", txID)
-	defer c.metrics.EmitKey(0, "committer", "end", "IsFinal", txID)
+func (c *Committer) IsFinal(ctx context.Context, txID string) error {
+	c.metrics.EmitKey(0, "Committer", "start", "IsFinal", txID)
+	defer c.metrics.EmitKey(0, "Committer", "end", "IsFinal", txID)
 
 	if logger.IsEnabledFor(zapcore.DebugLevel) {
 		logger.Debugf("Is [%s] final?", txID)
@@ -217,7 +220,7 @@ func (c *committer) IsFinal(ctx context.Context, txID string) error {
 				}
 				time.Sleep(100 * time.Millisecond)
 			default:
-				panic(fmt.Sprintf("invalid status code, got %c", vd))
+				return errors.Errorf("invalid status code, got [%c]", vd)
 			}
 		} else {
 			logger.Errorf("Is [%s] final? Failed getting transaction status from vault", txID)
@@ -229,7 +232,7 @@ func (c *committer) IsFinal(ctx context.Context, txID string) error {
 	return c.listenTo(ctx, txID, c.waitForEventTimeout)
 }
 
-func (c *committer) addListener(txid string, ch chan TxEvent) {
+func (c *Committer) addListener(txid string, ch chan TxEvent) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
@@ -242,7 +245,7 @@ func (c *committer) addListener(txid string, ch chan TxEvent) {
 	c.listeners[txid] = ls
 }
 
-func (c *committer) deleteListener(txid string, ch chan TxEvent) {
+func (c *Committer) deleteListener(txid string, ch chan TxEvent) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
@@ -259,7 +262,7 @@ func (c *committer) deleteListener(txid string, ch chan TxEvent) {
 	}
 }
 
-func (c *committer) notify(event TxEvent) {
+func (c *Committer) notify(event TxEvent) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
@@ -287,14 +290,13 @@ func (c *committer) notify(event TxEvent) {
 }
 
 // notifyChaincodeListeners notifies the chaincode event to the registered chaincode listeners.
-func (c *committer) notifyChaincodeListeners(event *ChaincodeEvent) error {
+func (c *Committer) notifyChaincodeListeners(event *ChaincodeEvent) {
 	c.publisher.Publish(event)
-	return nil
 }
 
-func (c *committer) listenTo(ctx context.Context, txid string, timeout time.Duration) error {
-	c.metrics.EmitKey(0, "committer", "start", "listenTo", txid)
-	defer c.metrics.EmitKey(0, "committer", "end", "listenTo", txid)
+func (c *Committer) listenTo(ctx context.Context, txid string, timeout time.Duration) error {
+	c.metrics.EmitKey(0, "Committer", "start", "listenTo", txid)
+	defer c.metrics.EmitKey(0, "Committer", "end", "listenTo", txid)
 
 	if logger.IsEnabledFor(zapcore.DebugLevel) {
 		logger.Debugf("Listen to finality of [%s]", txid)

platform/fabric/core/generic/committer/config.go

@@ -8,15 +8,16 @@ package committer
 
 import (
 	"github.com/hyperledger/fabric-protos-go/common"
+	"github.com/pkg/errors"
 )
 
-func (c *committer) handleConfig(block *common.Block, i int, env *common.Envelope) {
+func (c *Committer) handleConfig(block *common.Block, i int, env *common.Envelope) error {
 	committer, err := c.network.Committer(c.channel)
 	if err != nil {
-		logger.Panicf("Cannot get Committer [%s]", err)
+		return errors.Wrapf(err, "cannot get Committer for channel [%s]", c.channel)
 	}
-
 	if err := committer.CommitConfig(block.Header.Number, block.Data.Data[i], env); err != nil {
-		logger.Panicf("Cannot commit config envelope [%s]", err)
+		return errors.Wrapf(err, "cannot commit config envelope for channel [%s]", c.channel)
 	}
+	return nil
 }

platform/fabric/core/generic/committer/endorsertx.go

@@ -17,46 +17,47 @@ import (
 
 type ValidationFlags []uint8
 
-func (c *committer) handleEndorserTransaction(block *common.Block, i int, event *TxEvent, env *common.Envelope, chHdr *common.ChannelHeader) {
+func (c *Committer) handleEndorserTransaction(block *common.Block, i int, event *TxEvent, env *common.Envelope, chHdr *common.ChannelHeader) error {
 	txID := chHdr.TxId
 	event.Txid = txID
 
 	validationCode := pb.TxValidationCode(ValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])[i])
 	switch validationCode {
 	case pb.TxValidationCode_VALID:
 		if err := c.CommitEndorserTransaction(txID, block, i, env, event); err != nil {
-			logger.Panicf("failed committing transaction [%s] with err [%s]", txID, err)
+			return errors.Wrapf(err, "failed committing transaction [%s]", txID)
+		}
+		if err := c.getChaincodeEvents(env, block); err != nil {
+			return errors.Wrapf(err, "failed to publish chaincode events [%s]", txID)
 		}
-		c.getChaincodeEvents(env, block)
 	default:
 		if err := c.DiscardEndorserTransaction(txID, block, event, validationCode); err != nil {
-			logger.Panicf("failed discarding transaction [%s] with err [%s]", txID, err)
+			return errors.Wrapf(err, "failed discarding transaction [%s]", txID)
 		}
 	}
+	return nil
 }
 
 // getChaincodeEvents reads the chaincode events and notifies the listeners registered to the specific chaincode.
-func (c *committer) getChaincodeEvents(env *common.Envelope, block *common.Block) {
+func (c *Committer) getChaincodeEvents(env *common.Envelope, block *common.Block) error {
 	chaincodeEvent, err := readChaincodeEvent(env, block.Header.Number)
 	if err != nil {
-		logger.Panicf("error reading chaincode event", err)
+		return errors.Wrapf(err, "error reading chaincode event")
 	}
 	if chaincodeEvent != nil {
 		if logger.IsEnabledFor(zapcore.DebugLevel) {
 			logger.Debugf("Chaincode Event Received: ", chaincodeEvent)
 		}
-		err := c.notifyChaincodeListeners(chaincodeEvent)
-		if err != nil {
-			logger.Panicf("error sending chaincode events to listeners")
-		}
+		c.notifyChaincodeListeners(chaincodeEvent)
 	}
+	return nil
 }
 
 // CommitEndorserTransaction commits the transaction to the vault
-func (c *committer) CommitEndorserTransaction(txID string, block *common.Block, indexInBlock int, env *common.Envelope, event *TxEvent) error {
+func (c *Committer) CommitEndorserTransaction(txID string, block *common.Block, indexInBlock int, env *common.Envelope, event *TxEvent) error {
 	committer, err := c.network.Committer(c.channel)
 	if err != nil {
-		logger.Panicf("Cannot get Committer [%s]", err)
+		return errors.Wrapf(err, "cannot get Committer for channel [%s]", c.channel)
 	}
 
 	blockNum := block.Header.Number
@@ -70,7 +71,7 @@ func (c *committer) CommitEndorserTransaction(txID string, block *common.Block,
 
 	vc, deps, err := committer.Status(txID)
 	if err != nil {
-		logger.Panicf("failed getting tx's status [%s], with err [%s]", txID, err)
+		return errors.Wrapf(err, "failed getting tx's status [%s]", txID)
 	}
 	event.DependantTxIDs = append(event.DependantTxIDs, deps...)
 
@@ -88,23 +89,23 @@ func (c *committer) CommitEndorserTransaction(txID string, block *common.Block,
 	default:
 		if block != nil {
 			if err := committer.CommitTX(event.Txid, event.Block, event.IndexInBlock, env); err != nil {
-				logger.Panicf("failed committing transaction [%s] with deps [%v] with err [%s]", txID, deps, err)
+				return errors.Wrapf(err, "failed committing transaction [%s] with deps [%v]", txID, deps)
 			}
 			return nil
 		}
 
 		if err := committer.CommitTX(event.Txid, event.Block, event.IndexInBlock, nil); err != nil {
-			logger.Panicf("failed committing transaction [%s] with deps [%v] with err [%s]", txID, deps, err)
+			return errors.Wrapf(err, "failed committing transaction [%s] with deps [%v]", txID, deps)
 		}
 	}
 	return nil
 }
 
 // DiscardEndorserTransaction discards the transaction from the vault
-func (c *committer) DiscardEndorserTransaction(txID string, block *common.Block, event *TxEvent, validationCode pb.TxValidationCode) error {
+func (c *Committer) DiscardEndorserTransaction(txID string, block *common.Block, event *TxEvent, validationCode pb.TxValidationCode) error {
 	committer, err := c.network.Committer(c.channel)
 	if err != nil {
-		logger.Panicf("Cannot get Committer [%s]", err)
+		return errors.Wrapf(err, "cannot get Committer for channel [%s]", c.channel)
 	}
 
 	blockNum := block.Header.Number
@@ -114,7 +115,7 @@ func (c *committer) DiscardEndorserTransaction(txID string, block *common.Block,
 
 	vc, deps, err := committer.Status(txID)
 	if err != nil {
-		logger.Panicf("failed getting tx's status [%s], with err [%s]", txID, err)
+		return errors.Wrapf(err, "failed getting tx's status [%s]", txID)
 	}
 	event.DependantTxIDs = append(event.DependantTxIDs, deps...)
 	switch vc {
@@ -133,5 +134,6 @@ func (c *committer) DiscardEndorserTransaction(txID string, block *common.Block,
 			logger.Errorf("failed discarding tx in state db with err [%s]", err)
 		}
 	}
+
 	return nil
 }

platform/fabric/core/generic/committer/external.go

@@ -64,21 +64,21 @@ func (c *ExternalCommitter) Status(txid string) (fdriver.ValidationCode, []strin
 
 func (c *ExternalCommitter) Validate(txid string) (fdriver.ValidationCode, error) {
 	if c.c == nil {
-		panic("no external committer defined, programming error")
+		panic("no external Committer defined, programming error")
 	}
 	return c.c.Validate(txid)
 }
 
 func (c *ExternalCommitter) CommitTX(txid string, block uint64, indexInBloc int) error {
 	if c.c == nil {
-		panic("no external committer defined, programming error")
+		panic("no external Committer defined, programming error")
 	}
 	return c.c.CommitTX(txid, block, indexInBloc)
 }
 
 func (c *ExternalCommitter) DiscardTX(txid string) error {
 	if c.c == nil {
-		panic("no external committer defined, programming error")
+		panic("no external Committer defined, programming error")
 	}
 	return c.c.DiscardTX(txid)
 }
@@ -91,7 +91,7 @@ func GetExternalCommitter(name string, sp view2.ServiceProvider, vault driver.Va
 	}
 	c, err := drivers[dNames[0]].Open(name, sp, vault)
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed opening external committer [%s]", dNames[0])
+		return nil, errors.Wrapf(err, "failed opening external Committer [%s]", dNames[0])
 	}
 	return &ExternalCommitter{c: c}, nil
 }

platform/fabric/core/generic/delivery/delivery.go

@@ -36,6 +36,10 @@ var (
 	}
 )
 
+// Callback is the callback function prototype to alert the rest of the stack about the availability of a new block.
+// The function returns two argument a boolean to signal if delivery should be stopped, and an error
+// to signal an issue during the processing of the block.
+// In case of an error, the same block is re-processed after a delay.
 type Callback func(block *common.Block) (bool, error)
 
 // Vault models a key-value store that can be updated by committing rwsets
@@ -64,7 +68,7 @@ type Delivery struct {
 
 func New(channel string, sp view2.ServiceProvider, network Network, callback Callback, vault Vault, waitForEventTimeout time.Duration) (*Delivery, error) {
 	if len(channel) == 0 {
-		panic("expected a channel, got empty string")
+		return nil, errors.Errorf("expected a channel, got empty string")
 	}
 	d := &Delivery{
 		channel:             channel,
@@ -143,17 +147,9 @@ func (d *Delivery) Run(ctx context.Context) error {
 
 				stop, err := d.callback(r.Block)
 				if err != nil {
-					switch errors.Cause(err) {
-					case ErrComm:
-						logger.Errorf("error occurred when processing filtered block [%s], retry", err)
-						// retry
-						time.Sleep(10 * time.Second)
-						df = nil
-					default:
-						// Stop here
-						logger.Errorf("error occurred when processing filtered block [%s]", err)
-						return err
-					}
+					logger.Errorf("error occurred when processing filtered block [%s], retry...", err)
+					time.Sleep(10 * time.Second)
+					df = nil
 				}
 				if stop {
 					return nil

platform/fabric/core/generic/finality/fabric.go

@@ -41,7 +41,7 @@ type fabricFinality struct {
 
 func NewFabricFinality(channel string, network Network, hasher Hasher, waitForEventTimeout time.Duration) (*fabricFinality, error) {
 	if len(channel) == 0 {
-		panic("expected a channel, got empty string")
+		return nil, errors.Errorf("expected a channel, got empty string")
 	}
 
 	d := &fabricFinality{