Integration testing, fix misc bugs (#657)

Author: benbjohnson

cmd/litestream/main.go

@@ -20,6 +20,7 @@ import (
 	"github.com/benbjohnson/litestream/abs"
 	"github.com/benbjohnson/litestream/file"
 	"github.com/benbjohnson/litestream/gcs"
+	"github.com/benbjohnson/litestream/internal"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	_ "github.com/mattn/go-sqlite3"
@@ -157,6 +158,9 @@ type Config struct {
 	// Only includes L1 through the last non-snapshot level.
 	Levels []*CompactionLevelConfig `yaml:"levels"`
 
+	// Snapshot configuration
+	Snapshot SnapshotConfig `yaml:"snapshot"`
+
 	// List of databases to manage.
 	DBs []*DBConfig `yaml:"dbs"`
 
@@ -179,6 +183,12 @@ type Config struct {
 	ConfigPath string `yaml:"-"`
 }
 
+// SnapshotConfig configures snapshots.
+type SnapshotConfig struct {
+	Interval  time.Duration `yaml:"interval"`
+	Retention time.Duration `yaml:"retention"`
+}
+
 // LoggingConfig configures logging.
 type LoggingConfig struct {
 	Level  string `yaml:"level"`
@@ -278,12 +288,12 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 
 	logOptions := slog.HandlerOptions{
 		Level:       slog.LevelInfo,
-		ReplaceAttr: replaceAttr,
+		ReplaceAttr: internal.ReplaceAttr,
 	}
 
 	switch strings.ToUpper(config.Logging.Level) {
 	case "TRACE":
-		logOptions.Level = litestream.LevelTrace
+		logOptions.Level = internal.LevelTrace
 	case "DEBUG":
 		logOptions.Level = slog.LevelDebug
 	case "INFO":
@@ -308,13 +318,6 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	return config, nil
 }
 
-func replaceAttr(groups []string, a slog.Attr) slog.Attr {
-	if a.Key == slog.LevelKey && a.Value.Any().(slog.Level) == litestream.LevelTrace {
-		a.Value = slog.StringValue("TRACE")
-	}
-	return a
-}
-
 // CompactionLevelConfig the configuration for a single level of compaction.
 type CompactionLevelConfig struct {
 	Interval  time.Duration `yaml:"interval"`

cmd/litestream/replicate.go

@@ -116,6 +116,11 @@ func (c *ReplicateCommand) Run() (err error) {
 
 	levels := c.Config.CompactionLevels()
 	c.Store = litestream.NewStore(dbs, levels)
+	c.Store.SnapshotInterval = c.Config.Snapshot.Interval
+	c.Store.SnapshotRetention = c.Config.Snapshot.Retention
+	if err := c.Store.Open(context.Background()); err != nil {
+		return fmt.Errorf("cannot open store: %w", err)
+	}
 
 	// Notify user that initialization is done.
 	for _, db := range c.Store.DBs() {

db.go

@@ -467,7 +467,7 @@ func (db *DB) verifyHeadersMatch() error {
 	}
 	hdr1 := dec.Header()
 	if salt1 != hdr1.WALSalt1 || salt2 != hdr1.WALSalt2 {
-		db.Logger.Log(LevelTrace, "salt mismatch",
+		db.Logger.Log(internal.LevelTrace, "salt mismatch",
 			"path", ltxPath,
 			"wal", [2]uint32{salt1, salt2},
 			"ltx", [2]uint32{hdr1.WALSalt1, hdr1.WALSalt2})
@@ -662,7 +662,7 @@ func (db *DB) verify(ctx context.Context) (info syncInfo, err error) {
 	salt2 := binary.BigEndian.Uint32(hdr0[20:])
 
 	if salt1 != dec.Header().WALSalt1 || salt2 != dec.Header().WALSalt2 {
-		db.Logger.Log(ctx, LevelTrace, "wal restarted",
+		db.Logger.Log(ctx, internal.LevelTrace, "wal restarted",
 			"salt1", salt1,
 			"salt2", salt2)
 
@@ -698,7 +698,7 @@ func (db *DB) verify(ctx context.Context) (info syncInfo, err error) {
 	if ok, err := db.ltxDecoderContains(dec, pgno, buf[WALFrameHeaderSize:]); err != nil {
 		return info, fmt.Errorf("ltx contains: %w", err)
 	} else if !ok {
-		db.Logger.Log(ctx, LevelTrace, "cannot find last page in last ltx file", "pgno", pgno)
+		db.Logger.Log(ctx, internal.LevelTrace, "cannot find last page in last ltx file", "pgno", pgno, "offset", prevWALOffset)
 		info.reason = "last page does not exist in last ltx file, wal overwritten by another process"
 		return info, nil
 	}
@@ -709,8 +709,6 @@ func (db *DB) verify(ctx context.Context) (info syncInfo, err error) {
 }
 
 func (db *DB) ltxDecoderContains(dec *ltx.Decoder, pgno uint32, data []byte) (bool, error) {
-	println("dbg/ltxDecoderContains", pgno, "\n", internal.Hexdump(data))
-
 	buf := make([]byte, dec.Header().PageSize)
 	for {
 		var hdr ltx.PageHeader
@@ -721,11 +719,9 @@ func (db *DB) ltxDecoderContains(dec *ltx.Decoder, pgno uint32, data []byte) (bo
 		}
 
 		if pgno != hdr.Pgno {
-			println("dbg/ltxDecoderContains.ltx", pgno, "!=", hdr.Pgno)
 			continue
 		}
 		if !bytes.Equal(data, buf) {
-			println("dbg/ltxDecoderContains.data.mismatch\n", internal.Hexdump(data))
 			continue
 		}
 		return true, nil
@@ -750,12 +746,21 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 	txID := pos.TXID + 1
 
 	filename := db.LTXPath(0, txID, txID)
-	db.Logger.Debug("sync",
+
+	logArgs := []any{
 		"txid", txID.String(),
-		"chkpt", checkpointing,
-		"snap", info.snapshotting,
 		"offset", info.offset,
-		"reason", info.reason)
+	}
+	if checkpointing {
+		logArgs = append(logArgs, "chkpt", "true")
+	}
+	if info.snapshotting {
+		logArgs = append(logArgs, "snap", "true")
+	}
+	if info.reason != "" {
+		logArgs = append(logArgs, "reason", info.reason)
+	}
+	db.Logger.Debug("sync", logArgs...)
 
 	// Prevent internal checkpoints during sync. Ignore if already in a checkpoint.
 	if !checkpointing {
@@ -785,7 +790,8 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 		// If we cannot verify the previous frame
 		var pfmError *PrevFrameMismatchError
 		if rd, err = NewWALReaderWithOffset(walFile, info.offset, info.salt1, info.salt2, db.Logger); errors.As(err, &pfmError) {
-			db.Logger.Log(ctx, LevelTrace, "prev frame mismatch, snapshotting", "err", pfmError.Err)
+			db.Logger.Log(ctx, internal.LevelTrace, "prev frame mismatch, snapshotting", "err", pfmError.Err)
+			info.offset = WALHeaderSize
 			if rd, err = NewWALReader(walFile, db.Logger); err != nil {
 				return fmt.Errorf("new wal reader, after reset")
 			}
@@ -795,17 +801,23 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 	}
 
 	// Build a mapping of changed page numbers and their latest content.
-	pageMap, sz, walCommit, err := rd.PageMap(ctx)
+	pageMap, maxOffset, walCommit, err := rd.PageMap(ctx)
 	if err != nil {
 		return fmt.Errorf("page map: %w", err)
 	}
 	if walCommit > 0 {
 		commit = walCommit
 	}
 
+	var sz int64
+	if maxOffset > 0 {
+		sz = maxOffset - info.offset
+	}
+	assert(sz >= 0, fmt.Sprintf("wal size must be positive: sz=%d, maxOffset=%d, info.offset=%d", sz, maxOffset, info.offset))
+
 	// Exit if we have no new WAL pages and we aren't snapshotting.
 	if !info.snapshotting && sz == 0 {
-		db.Logger.Log(ctx, LevelTrace, "sync: skip", "reason", "no new wal pages")
+		db.Logger.Log(ctx, internal.LevelTrace, "sync: skip", "reason", "no new wal pages")
 		return nil
 	}
 
@@ -824,14 +836,15 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 	uid, gid := internal.Fileinfo(db.fileInfo)
 	_ = os.Chown(tmpFilename, uid, gid)
 
-	db.Logger.Log(ctx, LevelTrace, "encode header",
+	db.Logger.Log(ctx, internal.LevelTrace, "encode header",
 		"txid", txID.String(),
 		"commit", commit,
 		"walOffset", info.offset,
 		"walSize", sz,
 		"salt1", rd.salt1,
 		"salt2", rd.salt2)
 
+	timestamp := time.Now()
 	enc := ltx.NewEncoder(ltxFile)
 	if err := enc.EncodeHeader(ltx.Header{
 		Version:   ltx.Version,
@@ -840,7 +853,7 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 		Commit:    commit,
 		MinTXID:   txID,
 		MaxTXID:   txID,
-		Timestamp: time.Now().UnixMilli(),
+		Timestamp: timestamp.UnixMilli(),
 		WALOffset: info.offset,
 		WALSize:   sz,
 		WALSalt1:  rd.salt1,
@@ -876,9 +889,19 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 
 	// Atomically rename file to final path.
 	if err := os.Rename(tmpFilename, filename); err != nil {
+		delete(db.maxLTXFileInfos.m, 0) // clear cache if in unknown state
 		return fmt.Errorf("rename ltx file: %w", err)
 	}
 
+	// Update file info cache for L0.
+	db.maxLTXFileInfos.m[0] = &ltx.FileInfo{
+		Level:     0,
+		MinTXID:   txID,
+		MaxTXID:   txID,
+		CreatedAt: time.Now(),
+		Size:      enc.N(),
+	}
+
 	db.Logger.Debug("db sync", "status", "ok")
 
 	return nil
@@ -902,7 +925,7 @@ func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.
 
 		// If page exists in the WAL, read from there.
 		if offset, ok := pageMap[pgno]; ok {
-			db.Logger.Log(ctx, LevelTrace, "encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno)
+			db.Logger.Log(ctx, internal.LevelTrace, "encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno, "type", "db+wal")
 
 			if n, err := walFile.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
 				return fmt.Errorf("read page %d @ %d: %w", pgno, offset, err)
@@ -917,7 +940,7 @@ func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.
 		}
 
 		offset := int64(pgno-1) * int64(db.pageSize)
-		db.Logger.Log(ctx, LevelTrace, "encode page from database", "offset", offset, "pgno", pgno)
+		db.Logger.Log(ctx, internal.LevelTrace, "encode page from database", "offset", offset, "pgno", pgno)
 
 		// Otherwise read directly from the database file.
 		if _, err := db.f.ReadAt(data, offset); err != nil {
@@ -943,7 +966,7 @@ func (db *DB) writeLTXFromWAL(ctx context.Context, enc *ltx.Encoder, walFile *os
 	for _, pgno := range pgnos {
 		offset := pageMap[pgno]
 
-		db.Logger.Log(ctx, LevelTrace, "encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno)
+		db.Logger.Log(ctx, internal.LevelTrace, "encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno, "type", "walonly")
 
 		// Read source page using page map.
 		if n, err := walFile.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
@@ -1077,6 +1100,10 @@ func (db *DB) execCheckpoint(mode string) (err error) {
 
 // SnapshotReader returns the current position of the database & a reader that contains a full database snapshot.
 func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error) {
+	if db.pageSize == 0 {
+		return ltx.Pos{}, nil, fmt.Errorf("page size not initialized yet")
+	}
+
 	pos, err := db.Pos()
 	if err != nil {
 		return pos, nil, fmt.Errorf("pos: %w", err)
@@ -1113,7 +1140,7 @@ func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error) {
 		}
 
 		// Build a mapping of changed page numbers and their latest content.
-		pageMap, sz, walCommit, err := rd.PageMap(ctx)
+		pageMap, maxOffset, walCommit, err := rd.PageMap(ctx)
 		if err != nil {
 			pw.CloseWithError(fmt.Errorf("page map: %w", err))
 			return
@@ -1122,6 +1149,11 @@ func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error) {
 			commit = walCommit
 		}
 
+		var sz int64
+		if maxOffset > 0 {
+			sz = maxOffset - rd.Offset()
+		}
+
 		db.Logger.Debug("encode snapshot header",
 			"txid", pos.TXID.String(),
 			"commit", commit,

db_test.go

@@ -3,6 +3,7 @@ package litestream_test
 import (
 	"context"
 	"database/sql"
+	"flag"
 	"hash/crc64"
 	"log/slog"
 	"os"
@@ -12,9 +13,12 @@ import (
 	"time"
 
 	"github.com/benbjohnson/litestream"
+	"github.com/benbjohnson/litestream/internal"
 	"github.com/superfly/ltx"
 )
 
+var logLevel = flag.String("log.level", "debug", "")
+
 func TestDB_Path(t *testing.T) {
 	db := newDB(t, "/tmp/db")
 	if got, want := db.Path(), `/tmp/db`; got != want {
@@ -496,9 +500,16 @@ func TestDB_Snapshot(t *testing.T) {
 func newDB(tb testing.TB, path string) *litestream.DB {
 	tb.Helper()
 	tb.Logf("db=%s", path)
+
+	level := slog.LevelDebug
+	if strings.ToLower(*logLevel) == "trace" {
+		level = internal.LevelTrace
+	}
+
 	db := litestream.NewDB(path)
 	db.Logger = slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
-		Level: slog.LevelDebug,
+		Level:       level,
+		ReplaceAttr: internal.ReplaceAttr,
 	}))
 	return db
 }

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.19
 	github.com/pkg/sftp v1.13.6
 	github.com/prometheus/client_golang v1.17.0
-	github.com/superfly/ltx v0.3.16
+	github.com/superfly/ltx v0.3.17
 	golang.org/x/crypto v0.17.0
 	golang.org/x/sync v0.5.0
 	golang.org/x/sys v0.15.0

go.sum

@@ -166,8 +166,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/superfly/ltx v0.3.16 h1:osibxhNf/4+DOA4fKuPxXVMPKvqoHgtCdy0tYb+AdVQ=
-github.com/superfly/ltx v0.3.16/go.mod h1:Nf50QAIXU/ET4ua3AuQ2fh31MbgNQZA7r/DYx6Os77s=
+github.com/superfly/ltx v0.3.17 h1:7cFPsrPYxgfGlqyhyKQtE9tx6mLZmcBUOrCUx1J6ju8=
+github.com/superfly/ltx v0.3.17/go.mod h1:Nf50QAIXU/ET4ua3AuQ2fh31MbgNQZA7r/DYx6Os77s=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=

internal/internal.go

@@ -2,13 +2,16 @@ package internal
 
 import (
 	"io"
+	"log/slog"
 	"os"
 	"syscall"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
+const LevelTrace = slog.Level(slog.LevelDebug - 4)
+
 // ReadCloser wraps a reader to also attach a separate closer.
 type ReadCloser struct {
 	r io.Reader
@@ -127,6 +130,16 @@ func MkdirAll(path string, fi os.FileInfo) error {
 	return nil
 }
 
+func ReplaceAttr(groups []string, a slog.Attr) slog.Attr {
+	if a.Key == slog.LevelKey {
+		switch a.Value.Any() {
+		case LevelTrace:
+			a.Value = slog.StringValue("TRACE")
+		}
+	}
+	return a
+}
+
 // Shared replica metrics.
 var (
 	OperationTotalCounterVec = promauto.NewCounterVec(prometheus.CounterOpts{

litestream.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log/slog"
 	"os"
 	"path"
 	"path/filepath"
@@ -50,8 +49,6 @@ var (
 	LogFlags = 0
 )
 
-const LevelTrace = slog.Level(slog.LevelDebug - 4)
-
 func init() {
 	sql.Register("litestream-sqlite3", &sqlite3.SQLiteDriver{
 		ConnectHook: func(conn *sqlite3.SQLiteConn) error {