nsqd topic/channel: reset health on successful backend write

- channel and topic put reset ctx.nsqd.SetHealth
- only set if !IsHealthy; atomic load and flag check should be faster than mutex
- nsqd_test.go: change `exp` to `nexp` in `nequal`
- relates to #594

nsqd/channel.go

@@ -325,6 +325,8 @@ func (c *Channel) put(m *Message) error {
 				c.name, err)
 			c.ctx.nsqd.SetHealth(err)
 			return err
+		} else if !c.ctx.nsqd.IsHealthy() {
+			c.ctx.nsqd.SetHealth(nil)
 		}
 	}
 	return nil
@@ -571,7 +573,7 @@ func (c *Channel) messagePump() {
 		atomic.StoreInt32(&c.bufferedCount, 1)
 		c.clientMsgChan <- msg
 		atomic.StoreInt32(&c.bufferedCount, 0)
-		// the client will call back to mark as in-flight w/ it's info
+		// the client will call back to mark as in-flight w/ its info
 	}
 
 exit:

nsqd/channel_test.go

@@ -1,6 +1,9 @@
 package nsqd
 
 import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
 	"os"
 	"strconv"
 	"testing"
@@ -166,5 +169,59 @@ func TestChannelEmptyConsumer(t *testing.T) {
 		stats := cl.Stats()
 		equal(t, stats.InFlightCount, int64(0))
 	}
+}
 
+func TestChannelHealth(t *testing.T) {
+	opts := NewOptions()
+	opts.Logger = newTestLogger(t)
+	opts.MemQueueSize = 2
+
+	_, httpAddr, nsqd := mustStartNSQD(opts)
+	defer os.RemoveAll(opts.DataPath)
+	defer nsqd.Exit()
+
+	topic := nsqd.GetTopic("test")
+
+	channel := topic.GetChannel("channel")
+	// cause channel.messagePump to exit so we can set channel.backend without
+	// a data race. side effect is it closes clientMsgChan, and messagePump is
+	// neve restarted. note this isn't the intended usage of exitChan but gets
+	// around the data race.
+	channel.exitChan <- 1
+
+	channel = topic.GetChannel("channel")
+	channel.backend = &errorBackendQueue{}
+
+	msg := NewMessage(<-nsqd.idChan, make([]byte, 100))
+	err := channel.PutMessage(msg)
+	equal(t, err, nil)
+
+	msg = NewMessage(<-nsqd.idChan, make([]byte, 100))
+	err = channel.PutMessage(msg)
+	equal(t, err, nil)
+
+	msg = NewMessage(<-nsqd.idChan, make([]byte, 100))
+	err = channel.PutMessage(msg)
+	nequal(t, err, nil)
+
+	url := fmt.Sprintf("http://%s/ping", httpAddr)
+	resp, err := http.Get(url)
+	equal(t, err, nil)
+	equal(t, resp.StatusCode, 500)
+	body, _ := ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	equal(t, string(body), "NOK - never gonna happen")
+
+	channel.backend = &errorRecoveredBackendQueue{}
+
+	msg = NewMessage(<-nsqd.idChan, make([]byte, 100))
+	err = channel.PutMessage(msg)
+	equal(t, err, nil)
+
+	resp, err = http.Get(url)
+	equal(t, err, nil)
+	equal(t, resp.StatusCode, 200)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	equal(t, string(body), "OK")
 }

nsqd/nsqd_test.go

@@ -39,7 +39,7 @@ func equal(t *testing.T, act, exp interface{}) {
 func nequal(t *testing.T, act, exp interface{}) {
 	if reflect.DeepEqual(exp, act) {
 		_, file, line, _ := runtime.Caller(1)
-		t.Logf("\033[31m%s:%d:\n\n\texp: %#v\n\n\tgot: %#v\033[39m\n\n",
+		t.Logf("\033[31m%s:%d:\n\n\tnexp: %#v\n\n\tgot:  %#v\033[39m\n\n",
 			filepath.Base(file), line, exp, act)
 		t.FailNow()
 	}

nsqd/topic.go

@@ -196,6 +196,8 @@ func (t *Topic) put(m *Message) error {
 				t.name, err)
 			t.ctx.nsqd.SetHealth(err)
 			return err
+		} else if !t.ctx.nsqd.IsHealthy() {
+			t.ctx.nsqd.SetHealth(nil)
 		}
 	}
 	return nil

nsqd/topic_test.go

@@ -59,6 +59,10 @@ func (d *errorBackendQueue) Delete() error         { return nil }
 func (d *errorBackendQueue) Depth() int64          { return 0 }
 func (d *errorBackendQueue) Empty() error          { return nil }
 
+type errorRecoveredBackendQueue struct{ errorBackendQueue }
+
+func (d *errorRecoveredBackendQueue) Put([]byte) error { return nil }
+
 func TestHealth(t *testing.T) {
 	opts := NewOptions()
 	opts.Logger = newTestLogger(t)
@@ -93,6 +97,19 @@ func TestHealth(t *testing.T) {
 	body, _ := ioutil.ReadAll(resp.Body)
 	resp.Body.Close()
 	equal(t, string(body), "NOK - never gonna happen")
+
+	topic.backend = &errorRecoveredBackendQueue{}
+
+	msg = NewMessage(<-nsqd.idChan, make([]byte, 100))
+	err = topic.PutMessages([]*Message{msg})
+	equal(t, err, nil)
+
+	resp, err = http.Get(url)
+	equal(t, err, nil)
+	equal(t, resp.StatusCode, 200)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	equal(t, string(body), "OK")
 }
 
 func TestDeletes(t *testing.T) {