nsqd: fix new topic channel lookup/creation

nsqd/nsqd.go

@@ -275,7 +275,6 @@ func (n *NSQd) GetTopic(topicName string) *Topic {
 		// release our global nsqd lock, and switch to a more granular topic lock while we init our
 		// channels from lookupd. This blocks concurrent PutMessages to this topic.
 		t.Lock()
-		defer t.Unlock()
 		n.Unlock()
 		// if using lookupd, make a blocking call to get the topics, and immediately create them.
 		// this makes sure that any message received is buffered to the right channels
@@ -285,6 +284,18 @@ func (n *NSQd) GetTopic(topicName string) *Topic {
 				t.getOrCreateChannel(channelName)
 			}
 		}
+		t.Unlock()
+
+		// NOTE: I would prefer for this to only happen in topic.GetChannel() but we're special
+		// casing the code above so that we can control the locks such that it is impossible
+		// for a message to be written to a (new) topic while we're looking up channels
+		// from lookupd...
+		//
+		// update messagePump state
+		select {
+		case t.channelUpdateChan <- 1:
+		case <-t.exitChan:
+		}
 	}
 	return t
 }

nsqd/topic.go

@@ -60,12 +60,15 @@ func (t *Topic) GetChannel(channelName string) *Channel {
 	t.Lock()
 	channel, isNew := t.getOrCreateChannel(channelName)
 	t.Unlock()
+
 	if isNew {
+		// update messagePump state
 		select {
 		case t.channelUpdateChan <- 1:
 		case <-t.exitChan:
 		}
 	}
+
 	return channel
 }
 
@@ -112,6 +115,7 @@ func (t *Topic) DeleteExistingChannel(channelName string) error {
 	// (so that we dont leave any messages around)
 	channel.Delete()
 
+	// update messagePump state
 	select {
 	case t.channelUpdateChan <- 1:
 	case <-t.exitChan: