nsqd: message loss during topic/channel bootstrap from metadata

[michaelyou]

#1031

I am not sure will this introduce other bugs~

This is what the pr do：

Initialize topic with pause=1, so when start t.messagePump goroutine，it won't read msg out.
In the method GetTopic，after NewTopic return, check n.isLoading to see if it is loading from metadata file(restart). If not, unpause it immediately, else, unpause it when all channels are created in LoadMetadata .

NewTopic is only called by GetTopic， so the change should not affect other code.

---

[ploxiln]

Above, you'll see that paused state is stored in the topic metadata, so it does:

 if t.Paused {
 	topic.Pause()
}

So you can remove that now, but here only un-pause if the topic was not originally paused.

[michaelyou]

Thanks, i will correct it

[ploxiln]

I wonder if we can move this below the channels block, as an alternative to taking the topic lock to block PutMessages() while getting channels.

[michaelyou]

The channelMap in getOrCreateChannel needs the topic's lock . I am not sure if we can do this~

// this expects the caller to handle locking
func (t *Topic) getOrCreateChannel(channelName string) (*Channel, bool) {
    channel, ok := t.channelMap[channelName]

[ploxiln]

Right - you'd call t.GetChannel() instead. But this is still tentative supposition :)

[mreiferson]

I wonder if we should just create a NewTopicWithChannels function that allows this to be handled in the correct order before goroutines are launched.

[michaelyou]

@mreiferson Creating a NewTopicWithChannels looks like a big change. I modified my commit and does it look more natural now~

[mreiferson]

It will be a slightly bigger change, but I feel like this code is a little too clever and surprising. We could comment why the ordering is what it is, but I feel like I'd rather have slightly more code that's clear in its intent?

What do you think @ploxiln?

[ploxiln]

I think the LoadMetadata() unpause-after style is reasonably elegant, and maybe we could improve the locking/latency of GetTopic() with that style. A NewTopicWithChannels() would not work for GetTopic() (because the nsqd lock would have to be held while querying lookupd). But I don't have a strong opinion - just adding a NewTopicWithChannels() just for LoadMetadata() would work fine too.

[ploxiln]

It looks like this changed since I last viewed it - instead of initializing a Topic with paused: true in NewTopic(), it calls Topic.Pause() just before releasing the nsqd lock in GetTopic() iff nsqd.isLoading. It seems to me like that works too.

Anyway, I think this is a fine minimal fix-up, and further refactoring can be proposed in a separate PR. @mreiferson?

[mreiferson]

LGTM after adding these comments, thanks!

[mreiferson]

Let's add a comment above this line that reads:

// if we're creating a new topic in this process while we're loading from metadata, the topic may already
// have message data persisted on disk. To ensure that all channels are created before message flow
// begins, we pause the topic.

[mreiferson]

And a comment above this line:

// this logic is reversed, and done _after_ channel creation to ensure that all channels
// are created before messages begin flowing

[michaelyou]

@mreiferson done

[ploxiln]

Thanks for squashing down to one commit. The final little issue is that the commit title is not descriptive of the overall change. I suggest:

nsqd: pause topics while loading metadata

[michaelyou]

@ploxiln OK, done

[ploxiln]

Thanks!