Merge pull request #1564 from rqlite/restore-only-on-empty

Only auto-restore if node is "empty"

CHANGELOG.md

@@ -8,6 +8,7 @@
 - [PR #1556](https://github.com/rqlite/rqlite/pull/1556): Add fast-path backup.
 - [PR #1556](https://github.com/rqlite/rqlite/pull/1556): rqlite CLI streams backup to file.
 - [PR #1557](https://github.com/rqlite/rqlite/pull/1557): Remove restriction on restores using SQLite files in WAL mode.
+- [PR #1564](https://github.com/rqlite/rqlite/pull/1564): Only auto-restore if the node is "empty". Fixes issue [#1561](https://github.com/rqlite/rqlite/issues/1561). Thanks @jtackaberry
 
 ## 8.14.1 (December 31st 2023)
 ### Implementation changes and bug fixes

cmd/rqlited/main.go

@@ -107,24 +107,32 @@ func main() {
 		log.Fatalf("failed to create store: %s", err.Error())
 	}
 
-	// Install the auto-restore file, if necessary.
+	// Install the auto-restore data, if necessary.
 	if cfg.AutoRestoreFile != "" {
-		log.Printf("auto-restore requested, initiating download")
-		start := time.Now()
-		path, errOK, err := restore.DownloadFile(mainCtx, cfg.AutoRestoreFile)
+		hd, err := store.HasData(str.Path())
 		if err != nil {
-			var b strings.Builder
-			b.WriteString(fmt.Sprintf("failed to download auto-restore file: %s", err.Error()))
-			if errOK {
-				b.WriteString(", continuing with node startup anyway")
-				log.Print(b.String())
-			} else {
-				log.Fatal(b.String())
-			}
+			log.Fatalf("failed to check for existing data: %s", err.Error())
+		}
+		if hd {
+			log.Printf("auto-restore requested, but data already exists in %s, skipping", str.Path())
 		} else {
-			log.Printf("auto-restore file downloaded in %s", time.Since(start))
-			if err := str.SetRestorePath(path); err != nil {
-				log.Fatalf("failed to preload auto-restore data: %s", err.Error())
+			log.Printf("auto-restore requested, initiating download")
+			start := time.Now()
+			path, errOK, err := restore.DownloadFile(mainCtx, cfg.AutoRestoreFile)
+			if err != nil {
+				var b strings.Builder
+				b.WriteString(fmt.Sprintf("failed to download auto-restore file: %s", err.Error()))
+				if errOK {
+					b.WriteString(", continuing with node startup anyway")
+					log.Print(b.String())
+				} else {
+					log.Fatal(b.String())
+				}
+			} else {
+				log.Printf("auto-restore file downloaded in %s", time.Since(start))
+				if err := str.SetRestorePath(path); err != nil {
+					log.Fatalf("failed to preload auto-restore data: %s", err.Error())
+				}
 			}
 		}
 	}

log/log.go

@@ -71,6 +71,22 @@ func (l *Log) LastCommandIndex(fi, li uint64) (uint64, error) {
 	return 0, nil
 }
 
+// HasCommand returns whether the Raft log contains any Command log entries.
+func (l *Log) HasCommand() (bool, error) {
+	fi, li, err := l.Indexes()
+	if err != nil {
+		return false, err
+	}
+	if fi == 0 || li == 0 {
+		return false, nil
+	}
+	i, err := l.LastCommandIndex(fi, li)
+	if err != nil {
+		return false, err
+	}
+	return i != 0, nil
+}
+
 // SetAppliedIndex sets the AppliedIndex value.
 func (l *Log) SetAppliedIndex(index uint64) error {
 	return l.SetUint64([]byte(rqliteAppliedIndex), index)

log/log_test.go

@@ -40,6 +40,14 @@ func Test_LogNewEmpty(t *testing.T) {
 	if lci != 0 {
 		t.Fatalf("got wrong value for last command index of not empty log: %d", lci)
 	}
+
+	f, err := l.HasCommand()
+	if err != nil {
+		t.Fatalf("failed to get has command: %s", err)
+	}
+	if f {
+		t.Fatalf("got wrong value for has command of empty log: %v", f)
+	}
 }
 
 func Test_LogNewExistNotEmpty(t *testing.T) {
@@ -91,6 +99,14 @@ func Test_LogNewExistNotEmpty(t *testing.T) {
 		t.Fatalf("got wrong value for last command index of not empty log: %d", lci)
 	}
 
+	f, err := l.HasCommand()
+	if err != nil {
+		t.Fatalf("failed to get has command: %s", err)
+	}
+	if !f {
+		t.Fatalf("got wrong value for has command of non-empty log: %v", f)
+	}
+
 	if err := l.Close(); err != nil {
 		t.Fatalf("failed to close log: %s", err)
 	}
@@ -311,6 +327,14 @@ func Test_LogDeleteAll(t *testing.T) {
 	if li != 0 {
 		t.Fatalf("got wrong value for last index of empty log: %d", li)
 	}
+
+	f, err := l.HasCommand()
+	if err != nil {
+		t.Fatalf("failed to get has command: %s", err)
+	}
+	if f {
+		t.Fatalf("got wrong value for has command of empty log: %v", f)
+	}
 }
 
 func Test_LogLastCommandIndexNotExist(t *testing.T) {

store/store.go

@@ -79,6 +79,7 @@ var (
 )
 
 const (
+	snapshotsDirName           = "rsnapshots"
 	restoreScratchPattern      = "rqlite-restore-*"
 	bootScatchPattern          = "rqlite-boot-*"
 	backupScatchPattern        = "rqlite-backup-*"
@@ -313,6 +314,37 @@ func IsNewNode(raftDir string) bool {
 	return !pathExists(filepath.Join(raftDir, raftDBPath))
 }
 
+// HasData returns true if the given dir indiciates that at least one FSM entry
+// has been committed to the log. This is true is there are any snapshots, or
+// if there are any entries in the log of raft.LogCommand type. This function
+// will block if the Bolt database is already open.
+func HasData(dir string) (bool, error) {
+	if !dirExists(dir) {
+		return false, nil
+	}
+	sstr, err := snapshot.NewStore(filepath.Join(dir, snapshotsDirName))
+	if err != nil {
+		return false, nil
+	}
+	snaps, err := sstr.List()
+	if err != nil {
+		return false, nil
+	}
+	if len(snaps) > 0 {
+		return true, nil
+	}
+	logs, err := rlog.New(filepath.Join(dir, raftDBPath), false)
+	if err != nil {
+		return false, nil
+	}
+	defer logs.Close()
+	h, err := logs.HasCommand()
+	if err != nil {
+		return false, nil
+	}
+	return h, nil
+}
+
 // Config represents the configuration of the underlying Store.
 type Config struct {
 	DBConf *DBConfig   // The DBConfig object for this Store.
@@ -428,7 +460,7 @@ func (s *Store) Open() (retErr error) {
 
 	// Upgrade any pre-existing snapshots.
 	oldSnapshotDir := filepath.Join(s.raftDir, "snapshots")
-	snapshotDir := filepath.Join(s.raftDir, "rsnapshots")
+	snapshotDir := filepath.Join(s.raftDir, snapshotsDirName)
 	if err := snapshot.Upgrade(oldSnapshotDir, snapshotDir, s.logger); err != nil {
 		return fmt.Errorf("failed to upgrade snapshots: %s", err)
 	}

store/store_test.go

@@ -50,6 +50,49 @@ func Test_OpenStoreSingleNode(t *testing.T) {
 	}
 }
 
+func Test_SingleNodeStore_HasData(t *testing.T) {
+	s, ln := mustNewStore(t)
+	defer ln.Close()
+
+	h, err := HasData(s.raftDir)
+	if err != nil {
+		t.Fatalf("failed to check for data: %s", err.Error())
+	}
+	if h {
+		t.Fatalf("new store has data")
+	}
+
+	if err := s.Open(); err != nil {
+		t.Fatalf("failed to open single-node store: %s", err.Error())
+	}
+	if err := s.Bootstrap(NewServer(s.ID(), s.Addr(), true)); err != nil {
+		t.Fatalf("failed to bootstrap single-node store: %s", err.Error())
+	}
+	if _, err := s.WaitForLeader(10 * time.Second); err != nil {
+		t.Fatalf("Error waiting for leader: %s", err)
+	}
+
+	// Write some data.
+	er := executeRequestFromStrings([]string{
+		`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
+	}, false, false)
+	_, err = s.Execute(er)
+	if err != nil {
+		t.Fatalf("failed to execute on single node: %s", err.Error())
+	}
+
+	// Close the store to unblock the Bolt database.
+	s.Close(true)
+
+	h, err = HasData(s.raftDir)
+	if err != nil {
+		t.Fatalf("failed to check for data: %s", err.Error())
+	}
+	if !h {
+		t.Fatalf("store does not have data")
+	}
+}
+
 // Test_SingleNodeSQLitePath ensures that basic functionality works when the SQLite database path
 // is explicitly specificed.
 func Test_SingleNodeOnDiskSQLitePath(t *testing.T) {

system_test/e2e/auto_state.py

@@ -112,6 +112,63 @@ class TestAutoRestoreS3(unittest.TestCase):
     os.remove(compressed_tmp_file)
     delete_s3_object(access_key_id, secret_access_key_id, S3_BUCKET, path)
 
+  @unittest.skipUnless(env_present('RQLITE_S3_ACCESS_KEY'), "S3 credentials not available")
+  def test_skipped_if_data(self):
+    '''Test that automatic restores are skipped if the node has data'''
+
+    node = None
+    cfg = None
+    path = None
+
+    access_key_id = os.environ['RQLITE_S3_ACCESS_KEY']
+    secret_access_key_id = os.environ['RQLITE_S3_SECRET_ACCESS_KEY']
+
+    # Upload a test SQLite file to S3.
+    tmp_file = self.create_sqlite_file()
+    compressed_tmp_file = temp_file()
+    gzip_compress(tmp_file, compressed_tmp_file)
+
+    path = "restore/"+random_string(32)
+    upload_s3_object(access_key_id, secret_access_key_id, S3_BUCKET, path, compressed_tmp_file)
+
+    # Create the auto-restore config file
+    auto_restore_cfg = {
+      "version": 1,
+      "type": "s3",
+      "sub" : {
+         "access_key_id": access_key_id,
+         "secret_access_key": secret_access_key_id,
+         "region": S3_BUCKET_REGION,
+         "bucket": S3_BUCKET,
+         "path": path
+      }
+    }
+    cfg = write_random_file(json.dumps(auto_restore_cfg))
+
+    # Create a new node, write some data to it.
+    n0 = Node(RQLITED_PATH, '0')
+    n0.start()
+    n0.wait_for_ready()
+    n0.execute('CREATE TABLE bar (id INTEGER NOT NULL PRIMARY KEY, name TEXT)')
+    n0.stop()
+
+    # Create a new node, using the directory from the previous node, but check
+    # that data is not restored from S3, wiping out the existing data.
+    n1 = Node(RQLITED_PATH, '0', dir=n0.dir, auto_restore=cfg)
+    n1.start()
+    n1.wait_for_ready()
+    j = n1.query('SELECT * FROM bar')
+    self.assertEqual(j, d_("{'results': [{'types': ['integer', 'text'], 'columns': ['id', 'name']}]}"))
+    j = n1.query('SELECT * FROM foo')
+    self.assertEqual(j, d_("{'results': [{'error': 'no such table: foo'}]}"))
+
+    deprovision_node(n0)
+    deprovision_node(n1)
+    os.remove(cfg)
+    os.remove(tmp_file)
+    os.remove(compressed_tmp_file)
+    delete_s3_object(access_key_id, secret_access_key_id, S3_BUCKET, path)
+
 class TestAutoBackupS3(unittest.TestCase):
   @unittest.skipUnless(env_present('RQLITE_S3_ACCESS_KEY'), "S3 credentials not available")
   def test_no_compress(self):