Force truncation checkpoint if WAL becomes runaway

A failsafe to force WAL truncation if somehow between normal
checkpoints the WAL grows to extreme sizes.

Default is around 200 megabytes with a 4k page size.

Author: hifi

db.go

@@ -31,6 +31,7 @@ const (
 	DefaultCheckpointInterval = 1 * time.Minute
 	DefaultMinCheckpointPageN = 1000
 	DefaultMaxCheckpointPageN = 10000
+	DefaultTruncatePageN      = 500000
 )
 
 // MaxIndex is the maximum possible WAL index.
@@ -83,6 +84,16 @@ type DB struct {
 	// unbounded if there are always read transactions occurring.
 	MaxCheckpointPageN int
 
+	// Threshold of WAL size, in pages, before a forced truncation checkpoint.
+	// A forced truncation checkpoint will block new transactions and wait for
+	// existing transactions to finish before issuing a checkpoint and
+	// truncating the WAL.
+	//
+	// If zero, no truncates are forced. This can cause the WAL to grow
+	// unbounded if there's a sudden spike of changes between other
+	// checkpoints.
+	TruncatePageN int
+
 	// Time between automatic checkpoints in the WAL. This is done to allow
 	// more fine-grained WAL files so that restores can be performed with
 	// better precision.
@@ -104,6 +115,7 @@ func NewDB(path string) *DB {
 
 		MinCheckpointPageN: DefaultMinCheckpointPageN,
 		MaxCheckpointPageN: DefaultMaxCheckpointPageN,
+		TruncatePageN:      DefaultTruncatePageN,
 		CheckpointInterval: DefaultCheckpointInterval,
 		MonitorInterval:    DefaultMonitorInterval,
 	}
@@ -740,7 +752,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	}
 
 	// Synchronize real WAL with current shadow WAL.
-	newWALSize, err := db.syncWAL(info)
+	origWALSize, newWALSize, err := db.syncWAL(info)
 	if err != nil {
 		return fmt.Errorf("sync wal: %w", err)
 	}
@@ -749,7 +761,9 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	// If WAL size is greater than min threshold, attempt checkpoint.
 	var checkpoint bool
 	checkpointMode := CheckpointModePassive
-	if db.MaxCheckpointPageN > 0 && newWALSize >= calcWALSize(db.pageSize, db.MaxCheckpointPageN) {
+	if db.TruncatePageN > 0 && origWALSize >= calcWALSize(db.pageSize, db.TruncatePageN) {
+		checkpoint, checkpointMode = true, CheckpointModeTruncate
+	} else if db.MaxCheckpointPageN > 0 && newWALSize >= calcWALSize(db.pageSize, db.MaxCheckpointPageN) {
 		checkpoint, checkpointMode = true, CheckpointModeRestart
 	} else if newWALSize >= calcWALSize(db.pageSize, db.MinCheckpointPageN) {
 		checkpoint = true
@@ -908,29 +922,29 @@ type syncInfo struct {
 }
 
 // syncWAL copies pending bytes from the real WAL to the shadow WAL.
-func (db *DB) syncWAL(info syncInfo) (newSize int64, err error) {
+func (db *DB) syncWAL(info syncInfo) (origSize int64, newSize int64, err error) {
 	// Copy WAL starting from end of shadow WAL. Exit if no new shadow WAL needed.
-	newSize, err = db.copyToShadowWAL(info.shadowWALPath)
+	origSize, newSize, err = db.copyToShadowWAL(info.shadowWALPath)
 	if err != nil {
-		return newSize, fmt.Errorf("cannot copy to shadow wal: %w", err)
+		return origSize, newSize, fmt.Errorf("cannot copy to shadow wal: %w", err)
 	} else if !info.restart {
-		return newSize, nil // If no restart required, exit.
+		return origSize, newSize, nil // If no restart required, exit.
 	}
 
 	// Parse index of current shadow WAL file.
 	dir, base := filepath.Split(info.shadowWALPath)
 	index, err := ParseWALPath(base)
 	if err != nil {
-		return 0, fmt.Errorf("cannot parse shadow wal filename: %s", base)
+		return 0, 0, fmt.Errorf("cannot parse shadow wal filename: %s", base)
 	}
 
 	// Start a new shadow WAL file with next index.
 	newShadowWALPath := filepath.Join(dir, FormatWALPath(index+1))
 	newSize, err = db.initShadowWALFile(newShadowWALPath)
 	if err != nil {
-		return 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
+		return 0, 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
 	}
-	return newSize, nil
+	return origSize, newSize, nil
 }
 
 func (db *DB) initShadowWALFile(filename string) (int64, error) {
@@ -966,58 +980,64 @@ func (db *DB) initShadowWALFile(filename string) (int64, error) {
 	_ = os.Chown(filename, uid, gid)
 
 	// Copy as much shadow WAL as available.
-	newSize, err := db.copyToShadowWAL(filename)
+	_, newSize, err := db.copyToShadowWAL(filename)
 	if err != nil {
 		return 0, fmt.Errorf("cannot copy to new shadow wal: %w", err)
 	}
 	return newSize, nil
 }
 
-func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
+func (db *DB) copyToShadowWAL(filename string) (origWalSize int64, newSize int64, err error) {
 	Tracef("%s: copy-shadow: %s", db.path, filename)
 
 	r, err := os.Open(db.WALPath())
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 	defer r.Close()
 
+	fi, err := r.Stat()
+	if err != nil {
+		return 0, 0, err
+	}
+	origWalSize = frameAlign(fi.Size(), db.pageSize)
+
 	w, err := os.OpenFile(filename, os.O_RDWR, 0666)
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 	defer w.Close()
 
-	fi, err := w.Stat()
+	fi, err = w.Stat()
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 	origSize := frameAlign(fi.Size(), db.pageSize)
 
 	// Read shadow WAL header to determine byte order for checksum & salt.
 	hdr := make([]byte, WALHeaderSize)
 	if _, err := io.ReadFull(w, hdr); err != nil {
-		return 0, fmt.Errorf("read header: %w", err)
+		return 0, 0, fmt.Errorf("read header: %w", err)
 	}
 	hsalt0 := binary.BigEndian.Uint32(hdr[16:])
 	hsalt1 := binary.BigEndian.Uint32(hdr[20:])
 
 	bo, err := headerByteOrder(hdr)
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 
 	// Read previous checksum.
 	chksum0, chksum1, err := readLastChecksumFrom(w, db.pageSize)
 	if err != nil {
-		return 0, fmt.Errorf("last checksum: %w", err)
+		return 0, 0, fmt.Errorf("last checksum: %w", err)
 	}
 
 	// Seek to correct position on real wal.
 	if _, err := r.Seek(origSize, io.SeekStart); err != nil {
-		return 0, fmt.Errorf("real wal seek: %w", err)
+		return 0, 0, fmt.Errorf("real wal seek: %w", err)
 	} else if _, err := w.Seek(origSize, io.SeekStart); err != nil {
-		return 0, fmt.Errorf("shadow wal seek: %w", err)
+		return 0, 0, fmt.Errorf("shadow wal seek: %w", err)
 	}
 
 	// Read through WAL from last position to find the page of the last
@@ -1032,7 +1052,7 @@ func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
 			Tracef("%s: copy-shadow: break %s @ %d; err=%s", db.path, filename, offset, err)
 			break // end of file or partial page
 		} else if err != nil {
-			return 0, fmt.Errorf("read wal: %w", err)
+			return 0, 0, fmt.Errorf("read wal: %w", err)
 		}
 
 		// Read frame salt & compare to header salt. Stop reading on mismatch.
@@ -1063,7 +1083,7 @@ func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
 		newDBSize := binary.BigEndian.Uint32(frame[4:])
 		if newDBSize != 0 {
 			if _, err := buf.WriteTo(w); err != nil {
-				return 0, fmt.Errorf("write shadow wal: %w", err)
+				return 0, 0, fmt.Errorf("write shadow wal: %w", err)
 			}
 			buf.Reset()
 			lastCommitSize = offset
@@ -1072,15 +1092,15 @@ func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
 
 	// Sync & close.
 	if err := w.Sync(); err != nil {
-		return 0, err
+		return 0, 0, err
 	} else if err := w.Close(); err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 
 	// Track total number of bytes written to WAL.
 	db.totalWALBytesCounter.Add(float64(lastCommitSize - origSize))
 
-	return lastCommitSize, nil
+	return origWalSize, lastCommitSize, nil
 }
 
 // ShadowWALReader opens a reader for a shadow WAL file at a given position.
@@ -1249,7 +1269,7 @@ func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
 	}
 
 	// Copy shadow WAL before checkpoint to copy as much as possible.
-	if _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
+	if _, _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
 		return fmt.Errorf("cannot copy to end of shadow wal before checkpoint: %w", err)
 	}
 
@@ -1284,7 +1304,7 @@ func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
 	}
 
 	// Copy the end of the previous WAL before starting a new shadow WAL.
-	if _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
+	if _, _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
 		return fmt.Errorf("cannot copy to end of shadow wal: %w", err)
 	}
 
@@ -1343,7 +1363,7 @@ func (db *DB) execCheckpoint(mode string) (err error) {
 	if err := db.db.QueryRow(rawsql).Scan(&row[0], &row[1], &row[2]); err != nil {
 		return err
 	}
-	Tracef("%s: checkpoint: mode=%v (%d,%d,%d)", db.path, mode, row[0], row[1], row[2])
+	log.Printf("%s: checkpoint: mode=%v (%d,%d,%d)", db.path, mode, row[0], row[1], row[2])
 
 	// Reacquire the read lock immediately after the checkpoint.
 	if err := db.acquireReadLock(); err != nil {