nsqd: new metadata filename without ID

symlink old metadta filename to new
when loading, if both exist, ensure they match
this makes rollback possible without losing messages
(when rolling back forward, some manual intervention is required)

Author: ploxiln

apps/nsqd/nsqd.go

@@ -218,8 +218,11 @@ func (p *program) Start() error {
 	options.Resolve(opts, flagSet, cfg)
 	nsqd := nsqd.New(opts)
 
-	nsqd.LoadMetadata()
-	err := nsqd.PersistMetadata()
+	err := nsqd.LoadMetadata()
+	if err != nil {
+		log.Fatalf("ERROR: %s", err.Error())
+	}
+	err = nsqd.PersistMetadata()
 	if err != nil {
 		log.Fatalf("ERROR: failed to persist metadata - %s", err.Error())
 	}

nsqd/nsqd.go

@@ -1,6 +1,7 @@
 package nsqd
 
 import (
+	"bytes"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/json"
@@ -267,24 +268,51 @@ type meta struct {
 	} `json:"topics"`
 }
 
-func (n *NSQD) LoadMetadata() {
+func readOrEmpty(fn string) ([]byte, error) {
+	data, err := ioutil.ReadFile(fn)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			return nil, fmt.Errorf("failed to read metadata from %s - %s", fn, err)
+		}
+	}
+	return data, nil
+}
+
+func (n *NSQD) LoadMetadata() error {
 	atomic.StoreInt32(&n.isLoading, 1)
 	defer atomic.StoreInt32(&n.isLoading, 0)
 
-	fn := fmt.Sprintf(path.Join(n.getOpts().DataPath, "nsqd.%d.dat"), n.getOpts().ID)
-	data, err := ioutil.ReadFile(fn)
+	fn := path.Join(n.getOpts().DataPath, "nsqd.dat")
+	// old metadata filename with ID, maintained in parallel to enable roll-back
+	fnID := path.Join(n.getOpts().DataPath, fmt.Sprintf("nsqd.%d.dat", n.getOpts().ID))
+
+	data, err := readOrEmpty(fn)
 	if err != nil {
-		if !os.IsNotExist(err) {
-			n.logf("ERROR: failed to read channel metadata from %s - %s", fn, err)
+		return err
+	}
+	dataID, errID := readOrEmpty(fnID)
+	if errID != nil {
+		return errID
+	}
+
+	if data == nil && dataID == nil {
+		return nil // fresh start
+	}
+	if data != nil && dataID != nil {
+		if bytes.Compare(data, dataID) != 0 {
+			return fmt.Errorf("metadata in %s and %s do not match (delete one)", fn, fnID)
 		}
-		return
+	}
+	if data == nil {
+		// only old metadata file exists, use it
+		fn = fnID
+		data = dataID
 	}
 
 	var m meta
 	err = json.Unmarshal(data, &m)
 	if err != nil {
-		n.logf("ERROR: failed to parse metadata - %s", err)
-		return
+		return fmt.Errorf("failed to parse metadata in %s - %s", fn, err)
 	}
 
 	for _, t := range m.Topics {
@@ -308,12 +336,15 @@ func (n *NSQD) LoadMetadata() {
 			}
 		}
 	}
+	return nil
 }
 
 func (n *NSQD) PersistMetadata() error {
-	// persist metadata about what topics/channels we have
-	// so that upon restart we can get back to the same state
-	fileName := fmt.Sprintf(path.Join(n.getOpts().DataPath, "nsqd.%d.dat"), n.getOpts().ID)
+	// persist metadata about what topics/channels we have, across restarts
+	fileName := path.Join(n.getOpts().DataPath, "nsqd.dat")
+	// old metadata filename with ID, maintained in parallel to enable roll-back
+	fileNameID := path.Join(n.getOpts().DataPath, fmt.Sprintf("nsqd.%d.dat", n.getOpts().ID))
+
 	n.logf("NSQ: persisting topic/channel metadata to %s", fileName)
 
 	js := make(map[string]interface{})
@@ -370,6 +401,21 @@ func (n *NSQD) PersistMetadata() error {
 		return err
 	}
 
+	stat, err := os.Lstat(fileNameID)
+	if err != nil || (stat.Mode()&os.ModeSymlink) == 0 {
+		tmpFileNameID := fmt.Sprintf("%s.%d.tmp", fileNameID, rand.Int())
+
+		err = os.Symlink(fileName, tmpFileNameID)
+		if err != nil {
+			return err
+		}
+		err = os.Rename(tmpFileNameID, fileNameID)
+		if err != nil {
+			return err
+		}
+	}
+
+	// technically should fsync DataPath here
 	return nil
 }
 

nsqd/nsqd_test.go

@@ -24,7 +24,7 @@ const (
 )
 
 func getMetadata(n *NSQD) (*meta, error) {
-	fn := fmt.Sprintf(path.Join(n.getOpts().DataPath, "nsqd.%d.dat"), n.getOpts().ID)
+	fn := path.Join(n.getOpts().DataPath, "nsqd.dat")
 	data, err := ioutil.ReadFile(fn)
 	if err != nil {
 		return nil, err