mitigation for nsqio#28 possible data corruption if the metadata for …

…a queue was not synced on previous shutdown

diskqueue.go

@@ -473,6 +473,29 @@ func (d *diskQueue) retrieveMetaData() error {
 	d.nextReadFileNum = d.readFileNum
 	d.nextReadPos = d.readPos
 
+	// if the metadata was not sync'd at the last shutdown of nsqd
+	// then the actual file size might actually be larger than the writePos,
+	// in which case the safest thing to do is skip to the next file for
+	// writes, and let the reader salvage what it can from the messages in the
+	// diskqueue beyond the metadata's likely also stale readPos
+	fileName = d.fileName(d.writeFileNum)
+	fileInfo, err := os.Stat(fileName)
+	if err != nil {
+		return err
+	}
+	fileSize := fileInfo.Size()
+	if d.writePos < fileSize {
+		d.logf(WARN,
+			"DISKQUEUE(%s) %s metadata writePos %d < file size of %d, skipping to new file",
+			d.name, fileName, d.writePos, fileSize)
+		d.writeFileNum += 1
+		d.writePos = 0
+		if d.writeFile != nil {
+			d.writeFile.Close()
+			d.writeFile = nil
+		}
+	}
+
 	return nil
 }