Merge pull request #28 from kaleido-io/fix-conn-dl

Fix hang broadcasting to broken connection

Author: nguyer

internal/ws/wsserver.go

@@ -224,6 +224,10 @@ func (s *webSocketServer) processReplies() {
 
 func (s *webSocketServer) broadcastToConnections(connections []*webSocketConnection, message interface{}) {
 	for _, c := range connections {
-		c.broadcast <- message
+		select {
+		case c.broadcast <- message:
+		case <-c.closing:
+			log.L(s.ctx).Warnf("Connection %s closed while attempting to deliver reply", c.id)
+		}
 	}
 }

internal/ws/wsserver_test.go

@@ -369,3 +369,20 @@ func TestListenTopicClosing(t *testing.T) {
 		topic: "test",
 	})
 }
+
+func TestBroadcastClosing(t *testing.T) {
+
+	w, ts := newTestWebSocketServer()
+	defer ts.Close()
+	w.getTopic("test")
+
+	c := &webSocketConnection{
+		server:   w,
+		topics:   make(map[string]*webSocketTopic),
+		closing:  make(chan struct{}),
+		newTopic: make(chan bool),
+	}
+	close(c.closing)
+	// Check this doesn't block
+	c.server.broadcastToConnections([]*webSocketConnection{c}, "anything")
+}

pkg/fftm/api.go

@@ -83,13 +83,8 @@ func (m *manager) runAPIServer() {
 }
 
 func (m *manager) runDebugServer() {
-	var debugServer *http.Server
 	debugPort := config.GetInt(tmconfig.DebugPort)
-
 	defer func() {
-		if debugServer != nil {
-			_ = debugServer.Close()
-		}
 		close(m.debugServerDone)
 	}()
 
@@ -100,8 +95,8 @@ func (m *manager) runDebugServer() {
 		r.PathPrefix("/debug/pprof/symbol").HandlerFunc(pprof.Symbol)
 		r.PathPrefix("/debug/pprof/trace").HandlerFunc(pprof.Trace)
 		r.PathPrefix("/debug/pprof/").HandlerFunc(pprof.Index)
-		debugServer = &http.Server{Addr: fmt.Sprintf("localhost:%d", debugPort), Handler: r, ReadHeaderTimeout: 30 * time.Second}
+		m.debugServer = &http.Server{Addr: fmt.Sprintf("localhost:%d", debugPort), Handler: r, ReadHeaderTimeout: 30 * time.Second}
 		log.L(m.ctx).Debugf("Debug HTTP endpoint listening on localhost:%d", debugPort)
-		_ = debugServer.ListenAndServe()
+		_ = m.debugServer.ListenAndServe()
 	}
 }

pkg/fftm/api_test.go

@@ -22,8 +22,10 @@ import (
 	"testing"
 
 	"github.com/go-resty/resty/v2"
+	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-transaction-manager/internal/confirmations"
+	"github.com/hyperledger/firefly-transaction-manager/internal/tmconfig"
 	"github.com/hyperledger/firefly-transaction-manager/mocks/confirmationsmocks"
 	"github.com/hyperledger/firefly-transaction-manager/mocks/ffcapimocks"
 	"github.com/hyperledger/firefly-transaction-manager/pkg/apitypes"
@@ -148,7 +150,7 @@ func TestSendTransactionE2E(t *testing.T) {
 
 }
 
-func TestDeployTransactionE2E(t *testing.T) {
+func TestDeployTransactionE2EWithDebugServer(t *testing.T) {
 
 	txSent := make(chan struct{})
 
@@ -195,6 +197,7 @@ func TestDeployTransactionE2E(t *testing.T) {
 		return n.NotificationType == confirmations.NewTransaction
 	})).Return(nil)
 
+	config.Set(tmconfig.DebugPort, 0)
 	m.Start()
 
 	req := strings.NewReader(sampleDeployTX)

pkg/fftm/manager.go

@@ -18,6 +18,7 @@ package fftm
 
 import (
 	"context"
+	"net/http"
 	"sync"
 	"time"
 
@@ -87,6 +88,7 @@ type manager struct {
 	blockListenerDone       chan struct{}
 	started                 bool
 	apiServerDone           chan error
+	debugServer             *http.Server
 	debugServerDone         chan struct{}
 
 	policyLoopInterval time.Duration
@@ -199,6 +201,9 @@ func (m *manager) Close() {
 	m.cancelCtx()
 	if m.started {
 		m.started = false
+		if m.debugServer != nil {
+			m.debugServer.Close()
+		}
 		<-m.apiServerDone
 		<-m.policyLoopDone
 		<-m.blockListenerDone

pkg/fftm/policyloop.go

@@ -37,26 +37,34 @@ func (m *manager) policyLoop() {
 	ctx := log.WithLogField(m.ctx, "role", "policyloop")
 
 	for {
+		// Wait to be notified, or timeout to run
 		timer := time.NewTimer(m.policyLoopInterval)
 		select {
 		case <-m.inflightUpdate:
-			m.policyLoopCycle(ctx, false)
-		case <-m.inflightStale:
-			m.policyLoopCycle(ctx, true)
 		case <-timer.C:
-			m.policyLoopCycle(ctx, false)
 		case <-ctx.Done():
 			log.L(ctx).Infof("Receipt poller exiting")
 			return
 		}
+		// Pop whether we were marked stale
+		stale := false
+		select {
+		case <-m.inflightStale:
+			stale = true
+		default:
+		}
+		m.policyLoopCycle(ctx, stale)
 	}
 }
 
 func (m *manager) markInflightStale() {
+	// First mark that we're stale
 	select {
 	case m.inflightStale <- true:
 	default:
 	}
+	// Then ensure we queue a loop that picks up the stale marker
+	m.markInflightUpdate()
 }
 
 func (m *manager) markInflightUpdate() {
@@ -237,6 +245,7 @@ func (m *manager) execPolicy(ctx context.Context, pending *pendingState, syncDel
 				log.L(ctx).Errorf("Policy engine returned error for transaction %s reason=%s: %s", mtx.ID, reason, err)
 				m.addError(mtx, reason, err)
 			} else {
+				log.L(ctx).Debugf("Policy engine executed for tx %s (update=%d,status=%s,hash=%s)", mtx.ID, update, mtx.Status, mtx.TransactionHash)
 				if mtx.FirstSubmit != nil && pending.trackingTransactionHash != mtx.TransactionHash {
 					// If now submitted, add to confirmations manager for receipt checking
 					m.trackSubmittedTransaction(ctx, pending)
@@ -257,6 +266,7 @@ func (m *manager) execPolicy(ctx context.Context, pending *pendingState, syncDel
 			}
 			if completed {
 				pending.remove = true // for the next time round the loop
+				log.L(ctx).Errorf("Transaction %s marked complete (status=%s): %s", mtx.ID, mtx.Status, err)
 				m.markInflightStale()
 			}
 		case policyengine.UpdateDelete: