Remove multiple replica support

Author: benbjohnson

cmd/litestream/databases.go

@@ -5,7 +5,6 @@ import (
 	"flag"
 	"fmt"
 	"os"
-	"strings"
 	"text/tabwriter"
 )
 
@@ -43,15 +42,9 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 			return err
 		}
 
-		var replicaNames []string
-		for _, r := range db.Replicas {
-			replicaNames = append(replicaNames, r.Name())
-		}
-
 		fmt.Fprintf(w, "%s\t%s\n",
 			db.Path(),
-			strings.Join(replicaNames, ","),
-		)
+			db.Replica.Name())
 	}
 
 	return nil

cmd/litestream/ltx.go

@@ -18,7 +18,6 @@ type LTXCommand struct{}
 func (c *LTXCommand) Run(ctx context.Context, args []string) (err error) {
 	fs := flag.NewFlagSet("litestream-ltx", flag.ContinueOnError)
 	configPath, noExpandEnv := registerConfigFlag(fs)
-	replicaName := fs.String("replica", "", "replica name")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -28,7 +27,6 @@ func (c *LTXCommand) Run(ctx context.Context, args []string) (err error) {
 		return fmt.Errorf("too many arguments")
 	}
 
-	var db *litestream.DB
 	var r *litestream.Replica
 	if isURL(fs.Arg(0)) {
 		if *configPath != "" {
@@ -49,62 +47,46 @@ func (c *LTXCommand) Run(ctx context.Context, args []string) (err error) {
 		}
 
 		// Lookup database from configuration file by path.
-		if path, err := expand(fs.Arg(0)); err != nil {
+		path, err := expand(fs.Arg(0))
+		if err != nil {
 			return err
-		} else if dbc := config.DBConfig(path); dbc == nil {
+		}
+		dbc := config.DBConfig(path)
+		if dbc == nil {
 			return fmt.Errorf("database not found in config: %s", path)
-		} else if db, err = NewDBFromConfig(dbc); err != nil {
-			return err
 		}
 
-		// Filter by replica, if specified.
-		if *replicaName != "" {
-			if r = db.Replica(*replicaName); r == nil {
-				return fmt.Errorf("replica %q not found for database %q", *replicaName, db.Path())
-			}
+		db, err := NewDBFromConfig(dbc)
+		if err != nil {
+			return err
+		} else if db.Replica == nil {
+			return fmt.Errorf("database has no replica")
 		}
-	}
-
-	// Find WAL files by db or replica.
-	var replicas []*litestream.Replica
-	if r != nil {
-		replicas = []*litestream.Replica{r}
-	} else {
-		replicas = db.Replicas
+		r = db.Replica
 	}
 
 	// List all WAL files.
 	w := tabwriter.NewWriter(os.Stdout, 0, 8, 2, ' ', 0)
 	defer w.Flush()
 
-	fmt.Fprintln(w, "replica\tmin_txid\tmax_txid\tsize\tcreated")
-	for _, r := range replicas {
-		if err := func() error {
-			itr, err := r.Client.LTXFiles(ctx, 0)
-			if err != nil {
-				return err
-			}
-			defer itr.Close()
-
-			for itr.Next() {
-				info := itr.Item()
-
-				fmt.Fprintf(w, "%s\t%x\t%d\t%d\t%s\n",
-					r.Name(),
-					info.MinTXID,
-					info.MaxTXID,
-					info.Size,
-					info.CreatedAt.Format(time.RFC3339),
-				)
-			}
-			return itr.Close()
-		}(); err != nil {
-			r.Logger().Error("cannot fetch ltx files", "error", err)
-			continue
-		}
+	fmt.Fprintln(w, "min_txid\tmax_txid\tsize\tcreated")
+	itr, err := r.Client.LTXFiles(ctx, 0)
+	if err != nil {
+		return err
 	}
+	defer itr.Close()
+
+	for itr.Next() {
+		info := itr.Item()
 
-	return nil
+		fmt.Fprintf(w, "%x\t%d\t%d\t%s\n",
+			info.MinTXID,
+			info.MaxTXID,
+			info.Size,
+			info.CreatedAt.Format(time.RFC3339),
+		)
+	}
+	return itr.Close()
 }
 
 // Usage prints the help screen to STDOUT.

cmd/litestream/main.go

@@ -285,7 +285,8 @@ type DBConfig struct {
 	MinCheckpointPageN *int           `yaml:"min-checkpoint-page-count"`
 	MaxCheckpointPageN *int           `yaml:"max-checkpoint-page-count"`
 
-	Replicas []*ReplicaConfig `yaml:"replicas"`
+	Replica  *ReplicaConfig   `yaml:"replica"`
+	Replicas []*ReplicaConfig `yaml:"replicas"` // Deprecated
 }
 
 // NewDBFromConfig instantiates a DB based on a configuration.
@@ -318,22 +319,37 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 		db.MaxCheckpointPageN = *dbc.MaxCheckpointPageN
 	}
 
-	// Instantiate and attach replicas.
-	for _, rc := range dbc.Replicas {
-		r, err := NewReplicaFromConfig(rc, db)
-		if err != nil {
-			return nil, err
-		}
-		db.Replicas = append(db.Replicas, r)
+	// Instantiate and attach replica.
+	// v0.3.x and before supported multiple replicas but that was dropped to
+	// ensure there's a single remote data authority.
+	if dbc.Replica == nil && len(dbc.Replicas) == 0 {
+		return nil, fmt.Errorf("must specify replica for database")
+	} else if dbc.Replica != nil && len(dbc.Replicas) > 0 {
+		return nil, fmt.Errorf("cannot specify 'replica' and 'replicas' on a database")
+	} else if len(dbc.Replicas) > 1 {
+		return nil, fmt.Errorf("multiple replicas on a single database are no longer supported")
+	}
+
+	var rc *ReplicaConfig
+	if dbc.Replica != nil {
+		rc = dbc.Replica
+	} else {
+		rc = dbc.Replicas[0]
+	}
+
+	r, err := NewReplicaFromConfig(rc, db)
+	if err != nil {
+		return nil, err
 	}
+	db.Replica = r
 
 	return db, nil
 }
 
 // ReplicaConfig represents the configuration for a single replica in a database.
 type ReplicaConfig struct {
 	Type                   string         `yaml:"type"` // "file", "s3"
-	Name                   string         `yaml:"name"` // name of replica, optional.
+	Name                   string         `yaml:"name"` // Deprecated
 	Path                   string         `yaml:"path"`
 	URL                    string         `yaml:"url"`
 	Retention              *time.Duration `yaml:"retention"`

cmd/litestream/replicate.go

@@ -107,23 +107,22 @@ func (c *ReplicateCommand) Run() (err error) {
 
 	// Notify user that initialization is done.
 	for _, db := range c.DBs {
+		r := db.Replica
 		slog.Info("initialized db", "path", db.Path())
-		for _, r := range db.Replicas {
-			slog := slog.With("name", r.Name(), "type", r.Client.Type(), "sync-interval", r.SyncInterval)
-			switch client := r.Client.(type) {
-			case *file.ReplicaClient:
-				slog.Info("replicating to", "path", client.Path())
-			case *s3.ReplicaClient:
-				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "region", client.Region, "endpoint", client.Endpoint)
-			case *gcs.ReplicaClient:
-				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path)
-			case *abs.ReplicaClient:
-				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "endpoint", client.Endpoint)
-			case *sftp.ReplicaClient:
-				slog.Info("replicating to", "host", client.Host, "user", client.User, "path", client.Path)
-			default:
-				slog.Info("replicating to")
-			}
+		slog := slog.With("name", r.Name(), "type", r.Client.Type(), "sync-interval", r.SyncInterval)
+		switch client := r.Client.(type) {
+		case *file.ReplicaClient:
+			slog.Info("replicating to", "path", client.Path())
+		case *s3.ReplicaClient:
+			slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "region", client.Region, "endpoint", client.Endpoint)
+		case *gcs.ReplicaClient:
+			slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path)
+		case *abs.ReplicaClient:
+			slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "endpoint", client.Endpoint)
+		case *sftp.ReplicaClient:
+			slog.Info("replicating to", "host", client.Host, "user", client.User, "path", client.Path)
+		default:
+			slog.Info("replicating to")
 		}
 	}
 

cmd/litestream/restore.go

@@ -137,13 +137,7 @@ func (c *RestoreCommand) loadFromConfig(ctx context.Context, dbPath, configPath
 		return nil, errSkipDBExists
 	}
 
-	// Determine the appropriate replica to restore from,
-	r, err := db.CalcRestoreTarget(ctx, *opt)
-	if err != nil {
-		return nil, err
-	}
-
-	return r, nil
+	return db.Replica, nil
 }
 
 // Usage prints the help screen to STDOUT.

db.go

@@ -98,9 +98,9 @@ type DB struct {
 	// The timeout to wait for EBUSY from SQLite.
 	BusyTimeout time.Duration
 
-	// List of replicas for the database.
+	// Remote replica for the database.
 	// Must be set before calling Open().
-	Replicas []*Replica
+	Replica *Replica
 
 	// Where to send log messages, defaults to global slog with database epath.
 	Logger *slog.Logger
@@ -213,16 +213,6 @@ func (db *DB) DirInfo() os.FileInfo {
 	return db.dirInfo
 }
 
-// Replica returns a replica by name.
-func (db *DB) Replica(name string) *Replica {
-	for _, r := range db.Replicas {
-		if r.Name() == name {
-			return r
-		}
-	}
-	return nil
-}
-
 // Pos returns the current replication position of the database.
 func (db *DB) Pos() (ltx.Pos, error) {
 	minTXID, maxTXID, err := db.MaxLTX()
@@ -267,15 +257,6 @@ func (db *DB) Open() (err error) {
 		return fmt.Errorf("minimum checkpoint page count required")
 	}
 
-	// Validate that all replica names are unique.
-	m := make(map[string]struct{})
-	for _, r := range db.Replicas {
-		if _, ok := m[r.Name()]; ok {
-			return fmt.Errorf("duplicate replica name: %q", r.Name())
-		}
-		m[r.Name()] = struct{}{}
-	}
-
 	// Clear old temporary files that my have been left from a crash.
 	if err := removeTmpFiles(db.metaPath); err != nil {
 		return fmt.Errorf("cannot remove tmp files: %w", err)
@@ -304,13 +285,13 @@ func (db *DB) Close(ctx context.Context) (err error) {
 	}
 
 	// Ensure replicas perform a final sync and stop replicating.
-	for _, r := range db.Replicas {
+	if db.Replica != nil {
 		if db.db != nil {
-			if e := r.Sync(ctx); e != nil && err == nil {
+			if e := db.Replica.Sync(ctx); e != nil && err == nil {
 				err = e
 			}
 		}
-		r.Stop(true)
+		db.Replica.Stop(true)
 	}
 
 	// Release the read lock to allow other applications to handle checkpointing.
@@ -440,8 +421,8 @@ func (db *DB) init() (err error) {
 	// TODO(gen): Generate diff of current LTX snapshot and save as next LTX file.
 
 	// Start replication.
-	for _, r := range db.Replicas {
-		r.Start(db.ctx)
+	if db.Replica != nil {
+		db.Replica.Start(db.ctx)
 	}
 
 	return nil
@@ -1096,34 +1077,6 @@ func (db *DB) monitor() {
 	}
 }
 
-// CalcRestoreTarget returns a replica to restore from based on opt criteria.
-func (db *DB) CalcRestoreTarget(ctx context.Context, opt RestoreOptions) (*Replica, error) {
-	var target struct {
-		replica   *Replica
-		updatedAt time.Time
-	}
-
-	for _, r := range db.Replicas {
-		// Skip replica if it does not match filter.
-		if opt.ReplicaName != "" && r.Name() != opt.ReplicaName {
-			continue
-		}
-
-		updatedAt, err := r.CalcRestoreTarget(ctx, opt)
-		if err != nil {
-			return nil, err
-		}
-
-		// Use the latest replica if we have multiple candidates.
-		if !updatedAt.After(target.updatedAt) {
-			continue
-		}
-
-		target.replica, target.updatedAt = r, updatedAt
-	}
-	return target.replica, nil
-}
-
 // CRC64 returns a CRC-64 ISO checksum of the database and its current position.
 //
 // This function obtains a read lock so it prevents syncs from occurring until