From 37c185ef2a21dec6291bcaed3a85656b109ea6fd Mon Sep 17 00:00:00 2001
From: Bryan Boreham <bjboreham@gmail.com>
Date: Fri, 15 Sep 2017 10:19:48 +0000
Subject: [PATCH] Try to find/fix spurious delay in unit test of consul

---
 app/multitenant/consul_pipe_router.go         | 10 ++++--
 .../consul_pipe_router_internal_test.go       |  4 ++-
 .../mock_consul_client_internal_test.go       | 32 ++++++++++++-------
 3 files changed, 32 insertions(+), 14 deletions(-)

diff --git a/app/multitenant/consul_pipe_router.go b/app/multitenant/consul_pipe_router.go
index bc06d2e11..8441ec94d 100644
--- a/app/multitenant/consul_pipe_router.go
+++ b/app/multitenant/consul_pipe_router.go
@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -111,12 +112,14 @@ func NewConsulPipeRouter(client ConsulClient, prefix, advertise string, userIDer
 }
 
 func (pr *consulPipeRouter) Stop() {
+	log.Infof("pr.Stop() %s", pr.advertise)
 	close(pr.quit)
 	pr.wait.Wait()
 }
 
 func (pr *consulPipeRouter) actor() {
 	defer pr.wait.Done()
+	defer log.Infof("consulPipeRouter %s actor done", pr.advertise)
 	for {
 		select {
 		case f := <-pr.actorChan:
@@ -392,6 +395,7 @@ func newBridgeConnection(key, addr string, pipe xfer.Pipe) *bridgeConnection {
 
 func (bc *bridgeConnection) stop() {
 	log.Infof("%s: Stopping client bridge connection", bc.key)
+	debug.PrintStack()
 	bc.mtx.Lock()
 	bc.stopped = true
 	if bc.conn != nil {
@@ -421,10 +425,11 @@ func (bc *bridgeConnection) loop() {
 		bc.mtx.Unlock()
 
 		// connect to other pipes instance
+		log.Infof("loop calling %s", url)
 		conn, _, err := xfer.DialWS(wsDialer, url, http.Header{})
 		if err != nil {
 			log.Errorf("%s: Client bridge connection; Error connecting to %s: %v", bc.key, url, err)
-			time.Sleep(time.Second) // TODO backoff
+			time.Sleep(100 * time.Millisecond) // TODO backoff
 			continue
 		}
 
@@ -437,9 +442,10 @@ func (bc *bridgeConnection) loop() {
 		bc.conn = conn
 		bc.mtx.Unlock()
 
-		if err := bc.pipe.CopyToWebsocket(end, conn); err != nil && !xfer.IsExpectedWSCloseError(err) {
+		if err = bc.pipe.CopyToWebsocket(end, conn); err != nil && !xfer.IsExpectedWSCloseError(err) {
 			log.Errorf("%s: Client bridge connection; Error copying pipe to websocket: %v", bc.key, err)
 		}
+		log.Infof("loop copy ended %s", err)
 		conn.Close()
 	}
 }
diff --git a/app/multitenant/consul_pipe_router_internal_test.go b/app/multitenant/consul_pipe_router_internal_test.go
index 3ba8e4c07..533d18a79 100644
--- a/app/multitenant/consul_pipe_router_internal_test.go
+++ b/app/multitenant/consul_pipe_router_internal_test.go
@@ -167,11 +167,13 @@ func (pt *pipeTest) reconnectPipe(t *testing.T) {
 
 func TestPipeRouter(t *testing.T) {
 	var (
-		consul     = newMockConsulClient()
+		mockQuit   = make(chan struct{})
+		consul     = newMockConsulClient(mockQuit)
 		replicas   = 2
 		iterations = 10
 		pt         = pipeTest{}
 	)
+	defer close(mockQuit)
 
 	for i := 0; i < replicas; i++ {
 		pr := NewConsulPipeRouter(consul, "", fmt.Sprintf("127.0.0.1:44%02d", i), NoopUserIDer)
diff --git a/app/multitenant/mock_consul_client_internal_test.go b/app/multitenant/mock_consul_client_internal_test.go
index 237f76aa0..20f9e2acf 100644
--- a/app/multitenant/mock_consul_client_internal_test.go
+++ b/app/multitenant/mock_consul_client_internal_test.go
@@ -8,15 +8,18 @@ import (
 )
 
 type mockKV struct {
-	mtx  sync.Mutex
-	cond *sync.Cond
-	kvps map[string]*consul.KVPair
-	next uint64 // the next update will have this 'index in the the log'
+	mtx     sync.Mutex
+	cond    *sync.Cond
+	kvps    map[string]*consul.KVPair
+	next    uint64 // the next update will have this 'index in the the log'
+	quit    chan struct{}
+	stopped bool
 }
 
-func newMockConsulClient() ConsulClient {
+func newMockConsulClient(quit chan struct{}) ConsulClient {
 	m := mockKV{
 		kvps: map[string]*consul.KVPair{},
+		quit: quit,
 	}
 	m.cond = sync.NewCond(&m.mtx)
 	go m.loop()
@@ -39,10 +42,17 @@ func copyKVPair(in *consul.KVPair) *consul.KVPair {
 
 // periodic loop to wake people up, so they can honour timeouts
 func (m *mockKV) loop() {
-	for range time.Tick(1 * time.Second) {
-		m.mtx.Lock()
-		m.cond.Broadcast()
-		m.mtx.Unlock()
+	ticker := time.Tick(1 * time.Second)
+	for {
+		select {
+		case <-ticker:
+			m.mtx.Lock()
+			m.cond.Broadcast()
+			m.mtx.Unlock()
+		case <-m.quit:
+			m.stopped = true
+			return
+		}
 	}
 }
 
@@ -72,7 +82,7 @@ func (m *mockKV) Get(key string, q *consul.QueryOptions) (*consul.KVPair, *consu
 	if !ok {
 		return nil, nil, nil
 	}
-	for q.WaitIndex >= value.ModifyIndex {
+	for q.WaitIndex >= value.ModifyIndex && !m.stopped {
 		m.cond.Wait()
 	}
 	return copyKVPair(value), nil, nil
@@ -82,7 +92,7 @@ func (m *mockKV) List(prefix string, q *consul.QueryOptions) (consul.KVPairs, *c
 	m.mtx.Lock()
 	defer m.mtx.Unlock()
 	deadline := time.Now().Add(q.WaitTime)
-	for m.next <= q.WaitIndex && time.Now().Before(deadline) {
+	for m.next <= q.WaitIndex && time.Now().Before(deadline) && !m.stopped {
 		m.cond.Wait()
 	}
 	if time.Now().After(deadline) {