Repository: benbjohnson/litestream
Pull request #477 (2023-05-07 06:15:20)

Prevent checkpoints during snapshots
Use a mutex that is held during snapshots and only try locking it during checkpoints.

If the database is in high write pressure and taking a snapshot takes long it is possible for the automatic checkpointing to go from PASSIVE to RESTART.

When a RESTART checkpoint is issued SQLite will block new read transactions and wait for the existing ones to finish. However, during a snapshot Litestream keeps a persistent read transaction open until it finishes which will in turn create a deadlock situation for the checkpointer as the RESTART checkpoint will start blocking writers as long as the snapshot is being written out.

This failure condition doesn't break everything persistently but it will create an unfortunate persistent write lock for the application until the snapshot finishes.

hifi: This actually fixes #237 as we hit the same issue.

Patch:
diff --git a/db.go b/db.go
index b7a31eb..9a443a4 100644
--- a/db.go
+++ b/db.go
@@ -44,11 +44,13 @@ const BusyTimeout = 1 * time.Second
 type DB struct {
 	mu       sync.RWMutex
 	path     string        // part to database
+	metaPath string        // Path to the database metadata.
 	db       *sql.DB       // target database
 	f        *os.File      // long-running db file descriptor
 	rtx      *sql.Tx       // long running read transaction
 	pageSize int           // page size, in bytes
 	notify   chan struct{} // closes on WAL change
+	chkMu    sync.Mutex    // checkpoint lock
 
 	fileInfo os.FileInfo // db info cached during init
 	dirInfo  os.FileInfo // parent dir info cached during init
@@ -94,18 +96,26 @@ type DB struct {
 	// List of replicas for the database.
 	// Must be set before calling Open().
 	Replicas []*Replica
+
+	// Where to send log messages, defaults to log.Default()
+	Logger *log.Logger
 }
 
 // NewDB returns a new instance of DB for a given path.
 func NewDB(path string) *DB {
+	dir, file := filepath.Split(path)
+
 	db := &DB{
-		path:   path,
-		notify: make(chan struct{}),
+		path:     path,
+		metaPath: filepath.Join(dir, "."+file+MetaDirSuffix),
+		notify:   make(chan struct{}),
 
 		MinCheckpointPageN: DefaultMinCheckpointPageN,
 		MaxCheckpointPageN: DefaultMaxCheckpointPageN,
 		CheckpointInterval: DefaultCheckpointInterval,
 		MonitorInterval:    DefaultMonitorInterval,
+
+		Logger: log.Default(),
 	}
 
 	db.dbSizeGauge = dbSizeGaugeVec.WithLabelValues(db.path)
@@ -142,20 +152,24 @@ func (db *DB) WALPath() string {
 
 // MetaPath returns the path to the database metadata.
 func (db *DB) MetaPath() string {
-	dir, file := filepath.Split(db.path)
-	return filepath.Join(dir, "."+file+MetaDirSuffix)
+	return db.metaPath
+}
+
+// SetMetaPath sets the path to database metadata.
+func (db *DB) SetMetaPath(mp string) {
+	db.metaPath = mp
 }
 
 // GenerationNamePath returns the path of the name of the current generation.
 func (db *DB) GenerationNamePath() string {
-	return filepath.Join(db.MetaPath(), "generation")
+	return filepath.Join(db.metaPath, "generation")
 }
 
 // GenerationPath returns the path of a single generation.
 // Panics if generation is blank.
 func (db *DB) GenerationPath(generation string) string {
 	assert(generation != "", "generation name required")
-	return filepath.Join(db.MetaPath(), "generations", generation)
+	return filepath.Join(db.metaPath, "generations", generation)
 }
 
 // ShadowWALDir returns the path of the shadow wal directory.
@@ -278,7 +292,7 @@ func (db *DB) Open() (err error) {
 	}
 
 	// Clear old temporary files that my have been left from a crash.
-	if err := removeTmpFiles(db.MetaPath()); err != nil {
+	if err := removeTmpFiles(db.metaPath); err != nil {
 		return fmt.Errorf("cannot remove tmp files: %w", err)
 	}
 
@@ -441,13 +455,13 @@ func (db *DB) init() (err error) {
 	}
 
 	// Ensure meta directory structure exists.
-	if err := internal.MkdirAll(db.MetaPath(), db.dirInfo); err != nil {
+	if err := internal.MkdirAll(db.metaPath, db.dirInfo); err != nil {
 		return err
 	}
 
 	// If we have an existing shadow WAL, ensure the headers match.
 	if err := db.verifyHeadersMatch(); err != nil {
-		log.Printf("%s: init: cannot determine last wal position, clearing generation; %s", db.path, err)
+		db.Logger.Printf("%s: init: cannot determine last wal position, clearing generation; %s", db.path, err)
 		if err := os.Remove(db.GenerationNamePath()); err != nil && !os.IsNotExist(err) {
 			return fmt.Errorf("remove generation name: %w", err)
 		}
@@ -517,7 +531,7 @@ func (db *DB) cleanGenerations() error {
 		return err
 	}
 
-	dir := filepath.Join(db.MetaPath(), "generations")
+	dir := filepath.Join(db.metaPath, "generations")
 	fis, err := ioutil.ReadDir(dir)
 	if os.IsNotExist(err) {
 		return nil
@@ -647,7 +661,7 @@ func (db *DB) createGeneration() (string, error) {
 	generation := hex.EncodeToString(buf)
 
 	// Generate new directory.
-	dir := filepath.Join(db.MetaPath(), "generations", generation)
+	dir := filepath.Join(db.metaPath, "generations", generation)
 	if err := internal.MkdirAll(dir, db.dirInfo); err != nil {
 		return "", err
 	}
@@ -726,7 +740,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 		if info.generation, err = db.createGeneration(); err != nil {
 			return fmt.Errorf("create generation: %w", err)
 		}
-		log.Printf("%s: sync: new generation %q, %s", db.path, info.generation, info.reason)
+		db.Logger.Printf("%s: sync: new generation %q, %s", db.path, info.generation, info.reason)
 
 		// Clear shadow wal info.
 		info.shadowWALPath = db.ShadowWALPath(info.generation, 0)
@@ -1234,6 +1248,12 @@ func (db *DB) Checkpoint(ctx context.Context, mode string) (err error) {
 // checkpointAndInit performs a checkpoint on the WAL file and initializes a
 // new shadow WAL file.
 func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
+	// Try getting a checkpoint lock, will fail during snapshots.
+	if !db.chkMu.TryLock() {
+		return nil
+	}
+	defer db.chkMu.Unlock()
+
 	shadowWALPath, err := db.CurrentShadowWALPath(generation)
 	if err != nil {
 		return err
@@ -1365,7 +1385,7 @@ func (db *DB) monitor() {
 
 		// Sync the database to the shadow WAL.
 		if err := db.Sync(db.ctx); err != nil && !errors.Is(err, context.Canceled) {
-			log.Printf("%s: sync error: %s", db.path, err)
+			db.Logger.Printf("%s: sync error: %s", db.path, err)
 		}
 	}
 }
@@ -1469,6 +1489,19 @@ func (db *DB) CRC64(ctx context.Context) (uint64, Pos, error) {
 	return h.Sum64(), pos, nil
 }
 
+// BeginSnapshot takes an internal snapshot lock preventing checkpoints.
+//
+// When calling this the caller must also call EndSnapshot() once the snapshot
+// is finished.
+func (db *DB) BeginSnapshot() {
+	db.chkMu.Lock()
+}
+
+// EndSnapshot releases the internal snapshot lock that prevents checkpoints.
+func (db *DB) EndSnapshot() {
+	db.chkMu.Unlock()
+}
+
 // DefaultRestoreParallelism is the default parallelism when downloading WAL files.
 const DefaultRestoreParallelism = 8
 
diff --git a/replica.go b/replica.go
index ae21c77..b633123 100644
--- a/replica.go
+++ b/replica.go
@@ -72,6 +72,9 @@ type Replica struct {
 	// Encryption identities and recipients
 	AgeIdentities []age.Identity
 	AgeRecipients []age.Recipient
+
+	// The logger to send logging messages to. Defaults to log.Default()
+	Logger *log.Logger
 }
 
 func NewReplica(db *DB, name string) *Replica {
@@ -84,6 +87,7 @@ func NewReplica(db *DB, name string) *Replica {
 		Retention:              DefaultRetention,
 		RetentionCheckInterval: DefaultRetentionCheckInterval,
 		MonitorEnabled:         true,
+		Logger:                 log.Default(),
 	}
 
 	return r
@@ -459,6 +463,10 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 	r.muf.Lock()
 	defer r.muf.Unlock()
 
+	// Prevent checkpoints during snapshot.
+	r.db.BeginSnapshot()
+	defer r.db.EndSnapshot()
+
 	// Issue a passive checkpoint to flush any pages to disk before snapshotting.
 	if _, err := r.db.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE);`); err != nil {
 		return info, fmt.Errorf("pre-snapshot checkpoint: %w", err)
@@ -534,7 +542,7 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 		return info, err
 	}
 
-	log.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)
+	r.Logger.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)
 
 	return info, nil
 }
@@ -602,7 +610,7 @@ func (r *Replica) deleteSnapshotsBeforeIndex(ctx context.Context, generation str
 		if err := r.Client.DeleteSnapshot(ctx, info.Generation, info.Index); err != nil {
 			return fmt.Errorf("delete snapshot %s/%08x: %w", info.Generation, info.Index, err)
 		}
-		log.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
+		r.Logger.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
 	}
 
 	return itr.Close()
@@ -634,7 +642,7 @@ func (r *Replica) deleteWALSegmentsBeforeIndex(ctx context.Context, generation s
 	if err := r.Client.DeleteWALSegments(ctx, a); err != nil {
 		return fmt.Errorf("delete wal segments: %w", err)
 	}
-	log.Printf("%s(%s): wal segmented deleted before %s/%08x: n=%d", r.db.Path(), r.Name(), generation, index, len(a))
+	r.Logger.Printf("%s(%s): wal segmented deleted before %s/%08x: n=%d", r.db.Path(), r.Name(), generation, index, len(a))
 
 	return nil
 }
@@ -671,7 +679,7 @@ func (r *Replica) monitor(ctx context.Context) {
 
 		// Synchronize the shadow wal into the replication directory.
 		if err := r.Sync(ctx); err != nil {
-			log.Printf("%s(%s): monitor error: %s", r.db.Path(), r.Name(), err)
+			r.Logger.Printf("%s(%s): monitor error: %s", r.db.Path(), r.Name(), err)
 			continue
 		}
 	}
@@ -699,7 +707,7 @@ func (r *Replica) retainer(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := r.EnforceRetention(ctx); err != nil {
-				log.Printf("%s(%s): retainer error: %s", r.db.Path(), r.Name(), err)
+				r.Logger.Printf("%s(%s): retainer error: %s", r.db.Path(), r.Name(), err)
 				continue
 			}
 		}
@@ -721,7 +729,7 @@ func (r *Replica) snapshotter(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if _, err := r.Snapshot(ctx); err != nil && err != ErrNoGeneration {
-				log.Printf("%s(%s): snapshotter error: %s", r.db.Path(), r.Name(), err)
+				r.Logger.Printf("%s(%s): snapshotter error: %s", r.db.Path(), r.Name(), err)
 				continue
 			}
 		}
@@ -749,7 +757,7 @@ func (r *Replica) validator(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := r.Validate(ctx); err != nil {
-				log.Printf("%s(%s): validation error: %s", r.db.Path(), r.Name(), err)
+				r.Logger.Printf("%s(%s): validation error: %s", r.db.Path(), r.Name(), err)
 				continue
 			}
 		}
@@ -786,7 +794,7 @@ func (r *Replica) Validate(ctx context.Context) error {
 		ReplicaName: r.Name(),
 		Generation:  pos.Generation,
 		Index:       pos.Index - 1,
-		Logger:      log.New(os.Stderr, "", 0),
+		Logger:      r.Logger,
 	}); err != nil {
 		return fmt.Errorf("cannot restore: %w", err)
 	}
@@ -811,7 +819,7 @@ func (r *Replica) Validate(ctx context.Context) error {
 	if mismatch {
 		status = "mismatch"
 	}
-	log.Printf("%s(%s): validator: status=%s db=%016x replica=%016x pos=%s", db.Path(), r.Name(), status, chksum0, chksum1, pos)
+	r.Logger.Printf("%s(%s): validator: status=%s db=%016x replica=%016x pos=%s", db.Path(), r.Name(), status, chksum0, chksum1, pos)
 
 	// Validate checksums match.
 	if mismatch {
@@ -853,7 +861,7 @@ func (r *Replica) waitForReplica(ctx context.Context, pos Pos) error {
 		// Obtain current position of replica, check if past target position.
 		curr := r.Pos()
 		if curr.IsZero() {
-			log.Printf("%s(%s): validator: no replica position available", db.Path(), r.Name())
+			r.Logger.Printf("%s(%s): validator: no replica position available", db.Path(), r.Name())
 			continue
 		}
 
@@ -1008,7 +1016,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	// Ensure logger exists.
 	logger := opt.Logger
 	if logger == nil {
-		logger = log.New(ioutil.Discard, "", 0)
+		logger = r.Logger
 	}
 
 	logPrefix := r.Name()


Test patch:
diff --git a/cmd/litestream/main.go b/cmd/litestream/main.go
index f1440d3..3bb20c1 100644
--- a/cmd/litestream/main.go
+++ b/cmd/litestream/main.go
@@ -86,20 +86,17 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 		}
 
 		// Setup signal handler.
-		ctx, cancel := context.WithCancel(ctx)
 		signalCh := signalChan()
 
-		if err := c.Run(ctx); err != nil {
+		if err := c.Run(); err != nil {
 			return err
 		}
 
 		// Wait for signal to stop program.
 		select {
 		case err = <-c.execCh:
-			cancel()
 			fmt.Println("subprocess exited, litestream shutting down")
 		case sig := <-signalCh:
-			cancel()
 			fmt.Println("signal received, litestream shutting down")
 
 			if c.cmd != nil {
@@ -250,6 +247,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"`
+	MetaPath           *string        `yaml:"meta-path"`
 	MonitorInterval    *time.Duration `yaml:"monitor-interval"`
 	CheckpointInterval *time.Duration `yaml:"checkpoint-interval"`
 	MinCheckpointPageN *int           `yaml:"min-checkpoint-page-count"`
@@ -269,6 +267,9 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 	db := litestream.NewDB(path)
 
 	// Override default database settings if specified in configuration.
+	if dbc.MetaPath != nil {
+		db.SetMetaPath(*dbc.MetaPath)
+	}
 	if dbc.MonitorInterval != nil {
 		db.MonitorInterval = *dbc.MonitorInterval
 	}
diff --git a/cmd/litestream/replicate.go b/cmd/litestream/replicate.go
index 1973183..cd95d96 100644
--- a/cmd/litestream/replicate.go
+++ b/cmd/litestream/replicate.go
@@ -94,7 +94,7 @@ func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err e
 }
 
 // Run loads all databases specified in the configuration.
-func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
+func (c *ReplicateCommand) Run() (err error) {
 	// Display version information.
 	log.Printf("litestream %s", Version)
 
@@ -162,7 +162,7 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 			return fmt.Errorf("cannot parse exec command: %w", err)
 		}
 
-		c.cmd = exec.CommandContext(ctx, execArgs[0], execArgs[1:]...)
+		c.cmd = exec.Command(execArgs[0], execArgs[1:]...)
 		c.cmd.Env = os.Environ()
 		c.cmd.Stdout = os.Stdout
 		c.cmd.Stderr = os.Stderr