[FAB-9683] Handle cc launch registration failures

Registry holds LaunchState to track launching instead of a bare channel.
When launches complete, the embedded channel is closed. The error
managed by LaunchState can be used to determine if the launch failed or
succeeded.

Change-Id: Ib069c5752915a8839cf9af50ab94224345e4ffc3
Signed-off-by: Matthew Sykes <sykesmat@us.ibm.com>

Author: sykesm

core/chaincode/chaincode_suite_test.go

@@ -68,11 +68,6 @@ type ccpackage interface {
 	ccprovider.CCPackage
 }
 
-//go:generate counterfeiter -o mock/launch_registry.go --fake-name LaunchRegistry . launchRegistry
-type launchRegistry interface {
-	chaincode.LaunchRegistry
-}
-
 //go:generate counterfeiter -o mock/chaincode_stream.go --fake-name ChaincodeStream . chaincodeStream
 type chaincodeStream interface {
 	ccintf.ChaincodeStream
@@ -115,6 +110,11 @@ type peerLedger interface {
 
 // NOTE: These are getting generated into the "fake" package to avoid import cycles. We need to revisit this.
 
+//go:generate counterfeiter -o fake/launch_registry.go --fake-name LaunchRegistry . launchRegistry
+type launchRegistry interface {
+	chaincode.LaunchRegistry
+}
+
 //go:generate counterfeiter -o fake/message_handler.go --fake-name MessageHandler . messageHandler
 type messageHandler interface {
 	chaincode.MessageHandler

core/chaincode/chaincode_support.go

@@ -93,7 +93,9 @@ func NewChaincodeSupport(
 	return cs
 }
 
-// Launch will launch the chaincode if not running (if running return nil) and will wait for handler of the chaincode to get into ready state.
+// Launch starts executing chaincode if it is not already running. This method
+// blocks until the peer side handler gets into ready state or encounters a fatal
+// error. If the chaincode is already running, it simply returns.
 func (cs *ChaincodeSupport) Launch(ctx context.Context, cccid *ccprovider.CCContext, spec ccprovider.ChaincodeSpecGetter) error {
 	cname := cccid.GetCanonicalName()
 	if cs.HandlerRegistry.Handler(cname) != nil {

core/chaincode/chaincode_support_test.go

@@ -1028,7 +1028,7 @@ func TestStartAndWaitSuccess(t *testing.T) {
 	cccid := ccprovider.NewCCContext("testchannel", "testcc", "0", "landwtimertest_txid", false, nil, nil)
 
 	//actual test - everythings good
-	err := launcher.startAndWaitForReady(context.Background(), cccid, cds)
+	err := launcher.start(context.Background(), cccid, cds)
 	if err != nil {
 		t.Fatalf("expected success but failed with error %s", err)
 	}
@@ -1053,7 +1053,7 @@ func TestStartAndWaitTimeout(t *testing.T) {
 	cccid := ccprovider.NewCCContext("testchannel", "testcc", "0", "landwtimertest_txid", false, nil, nil)
 
 	//the actual test - timeout 1000 > 500
-	err := launcher.startAndWaitForReady(context.Background(), cccid, cds)
+	err := launcher.start(context.Background(), cccid, cds)
 	if err == nil {
 		t.Fatalf("expected error but succeeded")
 	}
@@ -1077,7 +1077,7 @@ func TestStartAndWaitLaunchError(t *testing.T) {
 	cccid := ccprovider.NewCCContext("testchannel", "testcc", "0", "landwtimertest_txid", false, nil, nil)
 
 	//actual test - container launch gives error
-	err := launcher.startAndWaitForReady(context.Background(), cccid, cds)
+	err := launcher.start(context.Background(), cccid, cds)
 	if err == nil {
 		t.Fatalf("expected error but succeeded")
 	}

core/chaincode/mock/launch_registry.go renamed to core/chaincode/fake/launch_registry.go

@@ -38,6 +38,7 @@ type ACLProvider interface {
 type Registry interface {
 	Register(*Handler) error
 	Ready(cname string)
+	Failed(cname string, err error)
 	Deregister(cname string) error
 }
 
@@ -286,7 +287,7 @@ func (h *Handler) serialSend(msg *pb.ChaincodeMessage) error {
 
 	var err error
 	if err = h.chatStream.Send(msg); err != nil {
-		err = errors.WithMessage(err, fmt.Sprintf("[%s] Error sending %s", shorttxid(msg.Txid), msg.Type))
+		err = errors.WithMessage(err, fmt.Sprintf("[%s] error sending %s", shorttxid(msg.Txid), msg.Type))
 		chaincodeLogger.Errorf("%+v", err)
 	}
 	return err
@@ -421,25 +422,26 @@ func (h *Handler) sendReady() error {
 	}
 
 	h.state = Ready
-	h.Registry.Ready(h.chaincodeID.Name)
 
 	chaincodeLogger.Debugf("Changed to state ready for chaincode %+v", h.chaincodeID)
 
 	return nil
 }
 
-// notifyDuringStartup will send ready on registration
-func (h *Handler) notifyDuringStartup(val bool) {
-	// TODO: FAB-9683
-	if !val {
-		chaincodeLogger.Errorf("failed to start %s", h.chaincodeID)
-		return
+// notifyRegistry will send ready on registration success and
+// update the launch state of the chaincode in the handler registry.
+func (h *Handler) notifyRegistry(err error) {
+	if err == nil {
+		err = h.sendReady()
 	}
 
-	err := h.sendReady()
 	if err != nil {
-		chaincodeLogger.Debugf("sendReady failed: %s", err)
+		h.Registry.Failed(h.chaincodeID.Name, err)
+		chaincodeLogger.Errorf("failed to start %s", h.chaincodeID)
+		return
 	}
+
+	h.Registry.Ready(h.chaincodeID.Name)
 }
 
 // handleRegister is invoked when chaincode tries to register.
@@ -456,7 +458,7 @@ func (h *Handler) HandleRegister(msg *pb.ChaincodeMessage) {
 	h.chaincodeID = chaincodeID
 	err = h.Registry.Register(h)
 	if err != nil {
-		h.notifyDuringStartup(false)
+		h.notifyRegistry(err)
 		return
 	}
 
@@ -466,8 +468,8 @@ func (h *Handler) HandleRegister(msg *pb.ChaincodeMessage) {
 
 	chaincodeLogger.Debugf("Got %s for chaincodeID = %s, sending back %s", pb.ChaincodeMessage_REGISTER, chaincodeID, pb.ChaincodeMessage_REGISTERED)
 	if err := h.serialSend(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_REGISTERED}); err != nil {
-		chaincodeLogger.Errorf("Error sending %s: %s", pb.ChaincodeMessage_REGISTERED, err)
-		h.notifyDuringStartup(false)
+		chaincodeLogger.Errorf("error sending %s: %s", pb.ChaincodeMessage_REGISTERED, err)
+		h.notifyRegistry(err)
 		return
 	}
 
@@ -476,7 +478,7 @@ func (h *Handler) HandleRegister(msg *pb.ChaincodeMessage) {
 	chaincodeLogger.Debugf("Changed state to established for %+v", h.chaincodeID)
 
 	// for dev mode this will also move to ready automatically
-	h.notifyDuringStartup(true)
+	h.notifyRegistry(nil)
 }
 
 func (h *Handler) Notify(msg *pb.ChaincodeMessage) {

core/chaincode/fake/registry.go

@@ -16,16 +16,34 @@ import (
 type HandlerRegistry struct {
 	allowUnsolicitedRegistration bool // from cs.userRunsCC
 
-	mutex     sync.Mutex               // lock covering handlers and launching
-	handlers  map[string]*Handler      // chaincode cname to associated handler
-	launching map[string]chan struct{} // launching chaincodes to ready chan
+	mutex     sync.Mutex              // lock covering handlers and launching
+	handlers  map[string]*Handler     // chaincode cname to associated handler
+	launching map[string]*LaunchState // launching chaincodes to LaunchState
+}
+
+type LaunchState struct {
+	done chan struct{}
+	err  error
+}
+
+func NewLaunchState() *LaunchState {
+	return &LaunchState{
+		done: make(chan struct{}),
+	}
+}
+
+func (l *LaunchState) Done() <-chan struct{} { return l.done }
+func (l *LaunchState) Err() error            { return l.err }
+func (l *LaunchState) Notify(err error) {
+	l.err = err
+	close(l.done)
 }
 
 // NewHandlerRegistry constructs a HandlerRegistry.
 func NewHandlerRegistry(allowUnsolicitedRegistration bool) *HandlerRegistry {
 	return &HandlerRegistry{
 		handlers:                     map[string]*Handler{},
-		launching:                    map[string]chan struct{}{},
+		launching:                    map[string]*LaunchState{},
 		allowUnsolicitedRegistration: allowUnsolicitedRegistration,
 	}
 }
@@ -48,19 +66,20 @@ func (r *HandlerRegistry) hasLaunched(chaincode string) bool {
 	return false
 }
 
-// Launching indicates that chaincode is being launched. The channel that
-// is returned will be closed when registration completes. An error will be
-// returned if chaincode launch processing has already been initated.
-func (r *HandlerRegistry) Launching(cname string) (<-chan struct{}, error) {
+// Launching indicates that chaincode is being launched. The LaunchState that
+// is returned provides mechanisms to determine when the operation has completed
+// and whether or not it failed. An error will be returned if chaincode launch
+// processing has already been initated.
+func (r *HandlerRegistry) Launching(cname string) (*LaunchState, error) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 	if r.hasLaunched(cname) {
 		return nil, errors.Errorf("chaincode %s has already been launched", cname)
 	}
 
-	registerCompleteCh := make(chan struct{})
-	r.launching[cname] = registerCompleteCh
-	return registerCompleteCh, nil
+	launchState := NewLaunchState()
+	r.launching[cname] = launchState
+	return launchState, nil
 }
 
 // Ready indicates that the chaincode registration has completed and the
@@ -69,10 +88,21 @@ func (r *HandlerRegistry) Ready(cname string) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
-	registerCompleteCh := r.launching[cname]
-	if registerCompleteCh != nil {
+	launchStatus := r.launching[cname]
+	if launchStatus != nil {
 		delete(r.launching, cname)
-		close(registerCompleteCh)
+		launchStatus.Notify(nil)
+	}
+}
+
+// Failed indicates that registration of a launched chaincode has failed.
+func (r *HandlerRegistry) Failed(cname string, err error) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
+	launchStatus := r.launching[cname]
+	if launchStatus != nil {
+		launchStatus.Notify(err)
 	}
 }