Implement live read replication

cmd/litestream/main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
 	"github.com/benbjohnson/litestream/gcs"
+	"github.com/benbjohnson/litestream/http"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	_ "github.com/mattn/go-sqlite3"
@@ -267,6 +268,7 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 // DBConfig represents the configuration for a single database.
 type DBConfig struct {
 	Path                 string         `yaml:"path"`
+	Upstream             UpstreamConfig `yaml:"upstream"`
 	MonitorDelayInterval *time.Duration `yaml:"monitor-delay-interval"`
 	CheckpointInterval   *time.Duration `yaml:"checkpoint-interval"`
 	MinCheckpointPageN   *int           `yaml:"min-checkpoint-page-count"`
@@ -289,6 +291,14 @@ func NewDBFromConfigWithPath(dbc *DBConfig, path string) (*litestream.DB, error)
 	// Initialize database with given path.
 	db := litestream.NewDB(path)
 
+	// Attach upstream HTTP client if specified.
+	if upstreamURL := dbc.Upstream.URL; upstreamURL != "" {
+		if dbc.Upstream.Path == "" {
+			return nil, fmt.Errorf("upstream path required")
+		}
+		db.StreamClient = http.NewClient(upstreamURL, dbc.Upstream.Path)
+	}
+
 	// Override default database settings if specified in configuration.
 	if dbc.MonitorDelayInterval != nil {
 		db.MonitorDelayInterval = *dbc.MonitorDelayInterval
@@ -315,6 +325,11 @@ func NewDBFromConfigWithPath(dbc *DBConfig, path string) (*litestream.DB, error)
 	return db, nil
 }
 
+type UpstreamConfig struct {
+	URL  string `yaml:"url"`
+	Path string `yaml:"path"`
+}
+
 // ReplicaConfig represents the configuration for a single replica in a database.
 type ReplicaConfig struct {
 	Type                   string         `yaml:"type"` // "file", "s3"

cmd/litestream/replicate.go

@@ -6,19 +6,16 @@ import (
 	"fmt"
 	"io"
 	"log"
-	"net"
-	"net/http"
-	_ "net/http/pprof"
 	"os"
 	"os/exec"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
 	"github.com/benbjohnson/litestream/gcs"
+	"github.com/benbjohnson/litestream/http"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	"github.com/mattn/go-shellwords"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 // ReplicateCommand represents a command that continuously replicates SQLite databases.
@@ -35,7 +32,8 @@ type ReplicateCommand struct {
 
 	Config Config
 
-	server *litestream.Server
+	server     *litestream.Server
+	httpServer *http.Server
 }
 
 // NewReplicateCommand returns a new instance of ReplicateCommand.
@@ -143,22 +141,12 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 		}
 	}
 
-	// Serve metrics over HTTP if enabled.
+	// Serve HTTP if enabled.
 	if c.Config.Addr != "" {
-		hostport := c.Config.Addr
-		if host, port, _ := net.SplitHostPort(c.Config.Addr); port == "" {
-			return fmt.Errorf("must specify port for bind address: %q", c.Config.Addr)
-		} else if host == "" {
-			hostport = net.JoinHostPort("localhost", port)
+		c.httpServer = http.NewServer(c.server, c.Config.Addr)
+		if err := c.httpServer.Open(); err != nil {
+			return fmt.Errorf("cannot start http server: %w", err)
 		}
-
-		log.Printf("serving metrics on http://%s/metrics", hostport)
-		go func() {
-			http.Handle("/metrics", promhttp.Handler())
-			if err := http.ListenAndServe(c.Config.Addr, nil); err != nil {
-				log.Printf("cannot start metrics server: %s", err)
-			}
-		}()
 	}
 
 	// Parse exec commands args & start subprocess.
@@ -183,10 +171,17 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-// Close closes all open databases.
+// Close closes the HTTP server & all open databases.
 func (c *ReplicateCommand) Close() (err error) {
-	if e := c.server.Close(); e != nil && err == nil {
-		err = e
+	if c.httpServer != nil {
+		if e := c.httpServer.Close(); e != nil && err == nil {
+			err = e
+		}
+	}
+	if c.server != nil {
+		if e := c.server.Close(); e != nil && err == nil {
+			err = e
+		}
 	}
 	return err
 }