Multi-level compaction (#650)

Author: benbjohnson

abs/replica_client.go

@@ -91,11 +91,11 @@ func (c *ReplicaClient) Init(ctx context.Context) (err error) {
 }
 
 // LTXFiles returns an iterator over all available LTX files.
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error) {
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
 	}
-	return newLTXFileIterator(ctx, c, level), nil
+	return newLTXFileIterator(ctx, c, level, seek), nil
 }
 
 // WriteLTXFile writes an LTX file to remote storage.
@@ -206,6 +206,7 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 type ltxFileIterator struct {
 	client *ReplicaClient
 	level  int
+	seek   ltx.TXID
 
 	ch     chan ltx.FileInfo
 	g      errgroup.Group
@@ -216,10 +217,11 @@ type ltxFileIterator struct {
 	err  error
 }
 
-func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int) *ltxFileIterator {
+func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int, seek ltx.TXID) *ltxFileIterator {
 	itr := &ltxFileIterator{
 		client: client,
 		level:  level,
+		seek:   seek,
 		ch:     make(chan ltx.FileInfo),
 	}
 
@@ -234,12 +236,16 @@ func (itr *ltxFileIterator) fetch() error {
 	defer close(itr.ch)
 
 	dir := litestream.LTXLevelDir(itr.client.Path, itr.level)
+	prefix := dir + "/"
+	if itr.seek != 0 {
+		prefix += itr.seek.String()
+	}
 
 	var marker azblob.Marker
 	for marker.NotDone() {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
-		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: dir + "/"})
+		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: prefix})
 		if err != nil {
 			return err
 		}

cmd/litestream/databases.go

@@ -35,7 +35,7 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 	w := tabwriter.NewWriter(os.Stdout, 0, 8, 2, ' ', 0)
 	defer w.Flush()
 
-	fmt.Fprintln(w, "path\treplicas")
+	fmt.Fprintln(w, "path\treplica")
 	for _, dbConfig := range config.DBs {
 		db, err := NewDBFromConfig(dbConfig)
 		if err != nil {
@@ -44,7 +44,7 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 
 		fmt.Fprintf(w, "%s\t%s\n",
 			db.Path(),
-			db.Replica.Name())
+			db.Replica.Client.Type())
 	}
 
 	return nil

cmd/litestream/ltx.go

@@ -70,7 +70,7 @@ func (c *LTXCommand) Run(ctx context.Context, args []string) (err error) {
 	defer w.Flush()
 
 	fmt.Fprintln(w, "min_txid\tmax_txid\tsize\tcreated")
-	itr, err := r.Client.LTXFiles(ctx, 0)
+	itr, err := r.Client.LTXFiles(ctx, 0, 0)
 	if err != nil {
 		return err
 	}

cmd/litestream/main.go

@@ -153,6 +153,10 @@ type Config struct {
 	// Bind address for serving metrics.
 	Addr string `yaml:"addr"`
 
+	// List of stages in a multi-level compaction.
+	// Only includes L1 through the last non-snapshot level.
+	Levels []*CompactionLevelConfig `yaml:"levels"`
+
 	// List of databases to manage.
 	DBs []*DBConfig `yaml:"dbs"`
 
@@ -194,6 +198,23 @@ func DefaultConfig() Config {
 	return Config{}
 }
 
+// CompactionLevels returns a full list of compaction levels include L0.
+func (c *Config) CompactionLevels() litestream.CompactionLevels {
+	levels := litestream.CompactionLevels{
+		{Level: 0},
+	}
+
+	for i, lvl := range c.Levels {
+		levels = append(levels, &litestream.CompactionLevel{
+			Level:     i + 1,
+			Interval:  lvl.Interval,
+			Retention: lvl.Retention,
+		})
+	}
+
+	return levels
+}
+
 // DBConfig returns database configuration by path.
 func (c *Config) DBConfig(path string) *DBConfig {
 	for _, dbConfig := range c.DBs {
@@ -275,6 +296,12 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	return config, nil
 }
 
+// CompactionLevelConfig the configuration for a single level of compaction.
+type CompactionLevelConfig struct {
+	Interval  time.Duration `yaml:"interval"`
+	Retention time.Duration `yaml:"retention"`
+}
+
 // DBConfig represents the configuration for a single database.
 type DBConfig struct {
 	Path               string         `yaml:"path"`
@@ -391,7 +418,7 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 	}
 
 	// Build replica.
-	r := litestream.NewReplica(db, c.Name)
+	r := litestream.NewReplica(db)
 	if v := c.Retention; v != nil {
 		r.Retention = *v
 	}
@@ -482,7 +509,7 @@ func newFileReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_
 }
 
 // newS3ReplicaClientFromConfig returns a new instance of s3.ReplicaClient built from config.
-func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *s3.ReplicaClient, err error) {
+func newS3ReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *s3.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for s3 replica")
@@ -545,7 +572,7 @@ func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *s
 }
 
 // newGCSReplicaClientFromConfig returns a new instance of gcs.ReplicaClient built from config.
-func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *gcs.ReplicaClient, err error) {
+func newGCSReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *gcs.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for gcs replica")
@@ -584,7 +611,7 @@ func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *
 }
 
 // newABSReplicaClientFromConfig returns a new instance of abs.ReplicaClient built from config.
-func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *abs.ReplicaClient, err error) {
+func newABSReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *abs.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for abs replica")
@@ -627,7 +654,7 @@ func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *
 }
 
 // newSFTPReplicaClientFromConfig returns a new instance of sftp.ReplicaClient built from config.
-func newSFTPReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *sftp.ReplicaClient, err error) {
+func newSFTPReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *sftp.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for sftp replica")

cmd/litestream/replicate.go

@@ -28,8 +28,8 @@ type ReplicateCommand struct {
 
 	Config Config
 
-	// List of managed databases specified in the config.
-	DBs []*litestream.DB
+	// Manages the set of databases & compaction levels.
+	Store *litestream.Store
 }
 
 func NewReplicateCommand() *ReplicateCommand {
@@ -92,24 +92,23 @@ func (c *ReplicateCommand) Run() (err error) {
 		slog.Error("no databases specified in configuration")
 	}
 
+	var dbs []*litestream.DB
 	for _, dbConfig := range c.Config.DBs {
 		db, err := NewDBFromConfig(dbConfig)
 		if err != nil {
 			return err
 		}
-
-		// Open database & attach to program.
-		if err := db.Open(); err != nil {
-			return err
-		}
-		c.DBs = append(c.DBs, db)
+		dbs = append(dbs, db)
 	}
 
+	levels := c.Config.CompactionLevels()
+	c.Store = litestream.NewStore(dbs, levels)
+
 	// Notify user that initialization is done.
-	for _, db := range c.DBs {
+	for _, db := range c.Store.DBs() {
 		r := db.Replica
 		slog.Info("initialized db", "path", db.Path())
-		slog := slog.With("name", r.Name(), "type", r.Client.Type(), "sync-interval", r.SyncInterval)
+		slog := slog.With("type", r.Client.Type(), "sync-interval", r.SyncInterval)
 		switch client := r.Client.(type) {
 		case *file.ReplicaClient:
 			slog.Info("replicating to", "path", client.Path())
@@ -166,13 +165,8 @@ func (c *ReplicateCommand) Run() (err error) {
 
 // Close closes all open databases.
 func (c *ReplicateCommand) Close() (err error) {
-	for _, db := range c.DBs {
-		if e := db.Close(context.Background()); e != nil {
-			db.Logger.Error("error closing db", "error", e)
-			if err == nil {
-				err = e
-			}
-		}
+	if e := c.Store.Close(); e != nil {
+		slog.Error("failed to close database", "error", e)
 	}
 	return err
 }

cmd/litestream/restore.go

@@ -107,7 +107,7 @@ func (c *RestoreCommand) loadFromURL(ctx context.Context, replicaURL string, ifD
 }
 
 // loadFromConfig returns a replica & updates the restore options from a DB reference.
-func (c *RestoreCommand) loadFromConfig(ctx context.Context, dbPath, configPath string, expandEnv, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
+func (c *RestoreCommand) loadFromConfig(_ context.Context, dbPath, configPath string, expandEnv, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
 	// Load configuration.
 	config, err := ReadConfigFile(configPath, expandEnv)
 	if err != nil {

compaction_level.go

@@ -0,0 +1,106 @@
+package litestream
+
+import (
+	"fmt"
+	"time"
+)
+
+// SnapshotLevel represents the level which full snapshots are held.
+const SnapshotLevel = 9
+
+// CompactionLevel represents a single part of a multi-level compaction.
+// Each level merges LTX files from the previous level into larger time granularities.
+type CompactionLevel struct {
+	// The numeric level. Must match the index in the list of levels.
+	Level int
+
+	// The frequency that the level is compacted from the previous level.
+	Interval time.Duration
+
+	// The duration that files in this level are stored.
+	Retention time.Duration
+}
+
+// PrevCompactionAt returns the time when the last compaction occurred.
+// Returns the current time if it is exactly a multiple of the level interval.
+func (lvl *CompactionLevel) PrevCompactionAt(now time.Time) time.Time {
+	return now.Truncate(lvl.Interval).UTC()
+}
+
+// NextCompactionAt returns the time until the next compaction occurs.
+// Returns the current time if it is exactly a multiple of the level interval.
+func (lvl *CompactionLevel) NextCompactionAt(now time.Time) time.Time {
+	return lvl.PrevCompactionAt(now).Add(lvl.Interval)
+}
+
+// CompactionLevels represents a sorted slice of non-snapshot compaction levels.
+type CompactionLevels []*CompactionLevel
+
+// Level returns the compaction level at the given index.
+// Returns an error if the index is a snapshot level or is out of bounds.
+func (a CompactionLevels) Level(level int) (*CompactionLevel, error) {
+	if level == SnapshotLevel {
+		return nil, fmt.Errorf("invalid argument, snapshot level")
+	}
+	if level < 0 || level > a.MaxLevel() {
+		return nil, fmt.Errorf("level out of bounds: %d", level)
+	}
+	return a[level], nil
+}
+
+// MaxLevel return the highest non-snapshot compaction level.
+func (a CompactionLevels) MaxLevel() int {
+	return len(a) - 1
+}
+
+// Validate returns an error if the levels are invalid.
+func (a CompactionLevels) Validate() error {
+	if len(a) == 0 {
+		return fmt.Errorf("at least one compaction level is required")
+	}
+
+	for i, lvl := range a {
+		if i != lvl.Level {
+			return fmt.Errorf("compaction level number out of order: %d, expected %d", lvl.Level, i)
+		} else if lvl.Level > SnapshotLevel-1 {
+			return fmt.Errorf("compaction level cannot exceed %d", SnapshotLevel-1)
+		}
+
+		if lvl.Level == 0 && lvl.Interval != 0 {
+			return fmt.Errorf("cannot set interval on compaction level zero")
+		}
+
+		if lvl.Level != 0 && lvl.Interval <= 0 {
+			return fmt.Errorf("interval required for level %d", lvl.Level)
+		}
+	}
+	return nil
+}
+
+// IsValidLevel returns true if level is a valid compaction level number.
+func (a CompactionLevels) IsValidLevel(level int) bool {
+	if level == SnapshotLevel {
+		return true
+	}
+	return level >= 0 && level < len(a)
+}
+
+// PrevLevel returns the previous compaction level.
+// Returns -1 if there is no previous level.
+func (a CompactionLevels) PrevLevel(level int) int {
+	if level == SnapshotLevel {
+		return a.MaxLevel()
+	}
+	return level - 1
+}
+
+// NextLevel returns the next compaction level.
+// Returns -1 if there is no next level.
+func (a CompactionLevels) NextLevel(level int) int {
+	if level == SnapshotLevel {
+		return -1
+	} else if level == a.MaxLevel() {
+		return SnapshotLevel
+	}
+	return level + 1
+}