Periodically check backup server for updates (#362)

Author: benbjohnson

cmd/litefs/config.go

@@ -53,6 +53,7 @@ func NewConfig() Config {
 	config.Lease.DemoteDelay = litefs.DefaultDemoteDelay
 
 	config.Backup.Delay = litefs.DefaultBackupDelay
+	config.Backup.FullSyncInterval = litefs.DefaultBackupFullSyncInterval
 
 	config.Log.Format = "text"
 
@@ -155,12 +156,13 @@ type LeaseConfig struct {
 
 // BackupConfig represents a config for backup services.
 type BackupConfig struct {
-	Type      string        `yaml:"type"`
-	Path      string        `yaml:"path"`       // "file" only
-	URL       string        `yaml:"url"`        // "litefs-cloud" only
-	Cluster   string        `yaml:"cluster"`    // "litefs-cloud" only
-	AuthToken string        `yaml:"auth-token"` // "litefs-cloud" only
-	Delay     time.Duration `yaml:"-"`
+	Type             string        `yaml:"type"`
+	Path             string        `yaml:"path"`       // "file" only
+	URL              string        `yaml:"url"`        // "litefs-cloud" only
+	Cluster          string        `yaml:"cluster"`    // "litefs-cloud" only
+	AuthToken        string        `yaml:"auth-token"` // "litefs-cloud" only
+	Delay            time.Duration `yaml:"-"`
+	FullSyncInterval time.Duration `yaml:"-"`
 }
 
 // LogConfig represents the configuration for logging.

cmd/litefs/mount_linux.go

@@ -378,6 +378,10 @@ func (c *MountCommand) initEnvironment(ctx context.Context) {
 func (c *MountCommand) initStoreBackupClient(ctx context.Context) error {
 	c.initBackupConfigFromEnv(ctx)
 
+	// Copy common fields.
+	c.Store.BackupDelay = c.Config.Backup.Delay
+	c.Store.BackupFullSyncInterval = c.Config.Backup.FullSyncInterval
+
 	switch typ := c.Config.Backup.Type; typ {
 	case "":
 		log.Printf("no backup client configured, skipping")

cmd/litefs/mount_test.go

@@ -357,7 +357,7 @@ func TestSingleNode_BackupClient(t *testing.T) {
 	t.Run("InitiateSnapshot", func(t *testing.T) {
 		cmd0 := newMountCommand(t, t.TempDir(), nil)
 		cmd0.Config.Data.Retention = 1 * time.Microsecond
-		cmd0.Config.Backup = main.BackupConfig{Type: "file", Path: t.TempDir()}
+		cmd0.Config.Backup = main.BackupConfig{Type: "file", Path: t.TempDir(), Delay: 1 * time.Millisecond}
 		runMountCommand(t, cmd0)
 
 		// Create a simple table with a single value.
@@ -394,8 +394,10 @@ func TestSingleNode_BackupClient(t *testing.T) {
 	// Ensure LiteFS can restore a database from backup if it doesn't exist locally.
 	t.Run("RestoreFromBackup/NoLocalDatabase", func(t *testing.T) {
 		backupDir := t.TempDir()
+		backupConfig := main.BackupConfig{Type: "file", Path: backupDir, Delay: 1 * time.Millisecond}
+
 		cmd0 := newMountCommand(t, t.TempDir(), nil)
-		cmd0.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd0.Config.Backup = backupConfig
 		runMountCommand(t, cmd0)
 
 		// Create a simple table with a single value.
@@ -416,7 +418,7 @@ func TestSingleNode_BackupClient(t *testing.T) {
 		t.Logf("shutdown original mount, starting new one")
 
 		cmd1 := newMountCommand(t, t.TempDir(), nil)
-		cmd1.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd1.Config.Backup = backupConfig
 		runMountCommand(t, cmd1)
 		waitForBackupSync(t, cmd1)
 
@@ -434,13 +436,14 @@ func TestSingleNode_BackupClient(t *testing.T) {
 	t.Run("RestoreFromBackup/RemoteAhead", func(t *testing.T) {
 		dir0, dir1 := t.TempDir(), t.TempDir()
 		backupDir := t.TempDir()
+		backupConfig := main.BackupConfig{Type: "file", Path: backupDir, Delay: 1 * time.Millisecond}
 
 		cmd0 := newMountCommand(t, dir0, nil)
-		cmd0.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd0.Config.Backup = backupConfig
 		waitForPrimary(t, runMountCommand(t, cmd0))
 
 		cmd1 := newMountCommand(t, dir1, cmd0)
-		cmd1.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd1.Config.Backup = backupConfig
 		runMountCommand(t, cmd1)
 
 		// Create a simple table with a single value.
@@ -479,7 +482,7 @@ func TestSingleNode_BackupClient(t *testing.T) {
 		// Restart the previous replica so it becomes the new primary.
 		t.Logf("restart replica and wait for promotion")
 		cmd1 = newMountCommand(t, dir1, nil)
-		cmd1.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd1.Config.Backup = backupConfig
 		waitForPrimary(t, runMountCommand(t, cmd1))
 		waitForBackupSync(t, cmd1)
 
@@ -498,13 +501,14 @@ func TestSingleNode_BackupClient(t *testing.T) {
 	t.Run("RestoreFromBackup/ChecksumMismatch", func(t *testing.T) {
 		dir0, dir1 := t.TempDir(), t.TempDir()
 		backupDir := t.TempDir()
+		backupConfig := main.BackupConfig{Type: "file", Path: backupDir, Delay: 1 * time.Millisecond}
 
 		cmd0 := newMountCommand(t, dir0, nil)
-		cmd0.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd0.Config.Backup = backupConfig
 		waitForPrimary(t, runMountCommand(t, cmd0))
 
 		cmd1 := newMountCommand(t, dir1, cmd0)
-		cmd1.Config.Backup = main.BackupConfig{Type: "file", Path: backupDir}
+		cmd1.Config.Backup = backupConfig
 		runMountCommand(t, cmd1)
 
 		// Create a simple table with a single value.
@@ -661,14 +665,85 @@ func TestSingleNode_BackupClient(t *testing.T) {
 		}
 
 		if got, want := cmd0.Config.Backup, (main.BackupConfig{
-			Type:      "litefs-cloud",
-			URL:       "https://litefs.fly.io",
-			AuthToken: "TOKENDATA",
-			Delay:     litefs.DefaultBackupDelay,
+			Type:             "litefs-cloud",
+			URL:              "https://litefs.fly.io",
+			AuthToken:        "TOKENDATA",
+			Delay:            litefs.DefaultBackupDelay,
+			FullSyncInterval: litefs.DefaultBackupFullSyncInterval,
 		}); got != want {
 			t.Fatalf("config=%#v, want %#v", got, want)
 		}
 	})
+
+	// Ensure LiteFS periodically syncs with the backup server to detect restores.
+	t.Run("FullSync", func(t *testing.T) {
+		backupConfig := main.BackupConfig{
+			Type:             "file",
+			Path:             t.TempDir(),
+			Delay:            10 * time.Microsecond,
+			FullSyncInterval: 10 * time.Millisecond,
+		}
+
+		cmd0 := newMountCommand(t, t.TempDir(), nil)
+		cmd0.Config.Backup = backupConfig
+		runMountCommand(t, cmd0)
+		waitForPrimary(t, cmd0)
+
+		// Create a simple table with a single value.
+		db0 := testingutil.OpenSQLDB(t, filepath.Join(cmd0.Config.FUSE.Dir, "db"))
+		if _, err := db0.Exec(`CREATE TABLE t (x)`); err != nil {
+			t.Fatal(err)
+		} else if _, err := db0.Exec(`INSERT INTO t VALUES (100)`); err != nil {
+			t.Fatal(err)
+		} else if err := db0.Close(); err != nil {
+			t.Fatal(err)
+		}
+
+		// Sync to backup.
+		if err := cmd0.Store.SyncBackup(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		// Start replica, sync & shutdown.
+		dir1 := t.TempDir()
+		cmd1 := newMountCommand(t, dir1, cmd0)
+		cmd1.Config.Backup = backupConfig
+		runMountCommand(t, cmd1)
+		waitForSync(t, "db", cmd0, cmd1)
+		if err := cmd1.Close(); err != nil {
+			t.Fatal(err)
+		}
+
+		// Restart replica as its own cluster so we can update the backup while
+		// the old primary is still live.
+		cmd1 = newMountCommand(t, dir1, nil)
+		cmd1.Config.Backup = backupConfig
+		runMountCommand(t, cmd1)
+		waitForPrimary(t, cmd1)
+
+		// Insert data into second primary.
+		db1 := testingutil.OpenSQLDB(t, filepath.Join(cmd1.Config.FUSE.Dir, "db"))
+		if _, err := db1.Exec(`INSERT INTO t VALUES (200)`); err != nil {
+			t.Fatal(err)
+		} else if err := db1.Close(); err != nil {
+			t.Fatal(err)
+		}
+		if err := cmd1.Store.SyncBackup(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+		waitForBackupSync(t, cmd1)
+
+		// Wait a moment for the first primary to pick up the changes.
+		time.Sleep(1 * time.Second)
+
+		db0 = testingutil.OpenSQLDB(t, filepath.Join(cmd1.Config.FUSE.Dir, "db"))
+		var n int
+		if err := db0.QueryRow(`SELECT SUM(x) FROM t`).Scan(&n); err != nil {
+			t.Fatal(err)
+		} else if got, want := n, 300; got != want {
+			t.Fatalf("sum=%d, want %d", got, want)
+		}
+	})
 }
 
 func TestMultiNode_Simple(t *testing.T) {

store.go

@@ -41,7 +41,8 @@ const (
 	DefaultHaltLockTTL             = 30 * time.Second
 	DefaultHaltLockMonitorInterval = 5 * time.Second
 
-	DefaultBackupDelay = 1 * time.Second
+	DefaultBackupDelay            = 1 * time.Second
+	DefaultBackupFullSyncInterval = 10 * time.Second
 )
 
 var ErrStoreClosed = fmt.Errorf("store closed")
@@ -103,6 +104,10 @@ type Store struct {
 	// This allows multiple changes in quick succession to be batched together.
 	BackupDelay time.Duration
 
+	// Interval between checks to re-fetch the position map. This ensures that
+	// restores on the backup server are detected by the LiteFS primary.
+	BackupFullSyncInterval time.Duration
+
 	// Callback to notify kernel of file changes.
 	Invalidator Invalidator
 
@@ -140,7 +145,8 @@ func NewStore(path string, candidate bool) *Store {
 		HaltLockTTL:             DefaultHaltLockTTL,
 		HaltLockMonitorInterval: DefaultHaltLockMonitorInterval,
 
-		BackupDelay: DefaultBackupDelay,
+		BackupDelay:            DefaultBackupDelay,
+		BackupFullSyncInterval: DefaultBackupFullSyncInterval,
 
 		Environment: &nopEnvironment{},
 	}
@@ -995,33 +1001,44 @@ func (s *Store) SyncBackup(ctx context.Context) error {
 }
 
 // streamBackup connects to a backup server and continuously streams LTX files.
-func (s *Store) streamBackup(ctx context.Context, oneTime bool) error {
+func (s *Store) streamBackup(ctx context.Context, oneTime bool) (err error) {
 	// Start subscription immediately so we can collect any changes.
 	subscription := s.Subscribe(0)
 	defer func() { _ = subscription.Close() }()
 
-	// Fetch position map from backup server.
-	posMap, err := s.BackupClient.PosMap(ctx)
-	if err != nil {
-		return fmt.Errorf("fetch position map: %w", err)
-	}
-
-	slog.Info("begin streaming backup", slog.Int("n", len(posMap)))
+	slog.Info("begin streaming backup", slog.Duration("full-sync-interval", s.BackupFullSyncInterval))
 	defer func() { slog.Info("exiting streaming backup") }()
 
-	// Build initial dirty set of databases.
-	dirtySet := make(map[string]struct{})
-	for name := range posMap {
-		dirtySet[name] = struct{}{}
-	}
-	for _, db := range s.DBs() {
-		dirtySet[db.Name()] = struct{}{}
-	}
-
 	timer := time.NewTimer(0)
 	defer timer.Stop()
 
+	var posMapTickerCh <-chan time.Time
+	if s.BackupFullSyncInterval > 0 {
+		ticker := time.NewTicker(s.BackupFullSyncInterval)
+		defer ticker.Stop()
+		posMapTickerCh = ticker.C
+	}
+
+	var posMap map[string]ltx.Pos
+	dirtySet := make(map[string]struct{})
+
+LOOP:
 	for {
+		// If we don't have a position map yet or if it's been reset then fetch
+		// a new one from the backup server and update our dirty set.
+		if posMap == nil {
+			slog.Debug("fetching position map from backup server")
+			if posMap, err = s.BackupClient.PosMap(ctx); err != nil {
+				return fmt.Errorf("fetch position map: %w", err)
+			}
+			for name := range posMap {
+				dirtySet[name] = struct{}{}
+			}
+			for _, db := range s.DBs() {
+				dirtySet[db.Name()] = struct{}{}
+			}
+		}
+
 		slog.Debug("syncing databases to backup", slog.Int("dirty", len(dirtySet)))
 
 		// Send pending transactions for each database.
@@ -1064,6 +1081,9 @@ func (s *Store) streamBackup(ctx context.Context, oneTime bool) error {
 		select {
 		case <-ctx.Done():
 			return nil
+		case <-posMapTickerCh: // periodically re-fetch position map from backup server
+			posMap = nil
+			continue LOOP
 		case <-subscription.NotifyCh():
 		}