litestream Documentation

go build -o bin/litestream ./cmd/litestream
go build -o bin/litestream-test ./cmd/litestream-test

./scripts/analyze-test-results.sh /tmp/litestream-overnight-<timestamp>

./scripts/setup-homebrew-tap.sh

ls /tmp/litestream-overnight-* -dt | head -1

./scripts/analyze-test-results.sh $(ls /tmp/litestream-overnight-* -dt | head -1)

LATEST=$(ls /tmp/litestream-* -dt | head -1)
tail -f $LATEST/logs/monitor.log
tail -f $LATEST/logs/litestream.log

ls -la bin/litestream bin/litestream-test

go build -o bin/litestream ./cmd/litestream
go build -o bin/litestream-test ./cmd/litestream-test

aws s3 ls s3://$S3_BUCKET/

echo $AWS_ACCESS_KEY_ID
echo $AWS_SECRET_ACCESS_KEY
echo $S3_BUCKET

grep -i error /tmp/litestream-*/logs/litestream.log | head -20

sqlite3 /tmp/litestream-*/test.db "SELECT COUNT(*) FROM test_data"
sqlite3 /tmp/litestream-*/restored.db "SELECT COUNT(*) FROM test_data"

/tmp/litestream-overnight-<timestamp>/
├── logs/
│   ├── litestream.log      # Litestream replication log
│   ├── load.log            # Load generator log
│   ├── monitor.log         # Real-time monitoring log
│   ├── populate.log        # Initial population log
│   └── validate.log        # Final validation log
├── test.db                 # Source database
├── test.db-wal             # Write-ahead log
├── test.db-shm             # Shared memory file
├── replica/                # Replica directory (file tests)
│   └── ltx/               # LTX files
└── restored.db            # Restored database for validation

cd basic
go run main.go

cd s3

# Set required environment variables
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export LITESTREAM_BUCKET="your-bucket-name"
export LITESTREAM_PATH="databases/myapp"  # optional, defaults to "litestream"
export AWS_REGION="us-east-1"             # optional, defaults to "us-east-1"

go run main.go

import (
    "context"
    "database/sql"
    "github.com/benbjohnson/litestream"
    "github.com/benbjohnson/litestream/file"  // or s3, gs, abs, etc.
    _ "modernc.org/sqlite"
)

// 1. Create database wrapper
db := litestream.NewDB("/path/to/db.sqlite")

// 2. Create replica client
client := file.NewReplicaClient("/path/to/replica")
// OR from URL:
// client, _ := litestream.NewReplicaClientFromURL("s3://bucket/path")

// 3. Attach replica to database
replica := litestream.NewReplicaWithClient(db, client)
db.Replica = replica
client.Replica = replica // file backend only; preserves ownership/permissions

// 4. Create compaction levels (L0 required, plus at least one more)
levels := litestream.CompactionLevels{
    {Level: 0},
    {Level: 1, Interval: 10 * time.Second},
}

// 5. Create Store to manage DB and background compaction
store := litestream.NewStore([]*litestream.DB{db}, levels)

// 6. Open Store (opens all DBs, starts background monitors)
if err := store.Open(ctx); err != nil { ... }
defer store.Close(context.Background())

// 7. Open your app's SQLite connection for normal database operations
// Use DSN params for PRAGMAs to ensure they apply to all connections in the pool
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(wal)", "/path/to/db.sqlite")
sqlDB, err := sql.Open("sqlite", dsn)
if err != nil { ... }

// Create replica without database for restore
replica := litestream.NewReplicaWithClient(nil, client)

opt := litestream.NewRestoreOptions()
opt.OutputPath = "/path/to/restored.db"
// Optional: point-in-time restore
// opt.Timestamp = time.Now().Add(-1 * time.Hour)

if err := replica.Restore(ctx, opt); err != nil {
    if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) {
        // No backup available, create fresh database
    }
    return err
}

store.SnapshotInterval  = 24 * time.Hour  // How often to create snapshots
store.SnapshotRetention = 24 * time.Hour  // How long to keep snapshots
store.L0Retention       = 5 * time.Minute // How long to keep L0 files after compaction

db.MonitorInterval    = 1 * time.Second   // How often to check for changes
db.CheckpointInterval = 1 * time.Minute   // Time-based checkpoint interval
db.MinCheckpointPageN = 1000              // Page threshold for checkpoint
db.BusyTimeout        = 1 * time.Second   // SQLite busy timeout

replica.SyncInterval   = 1 * time.Second  // Time between syncs
replica.MonitorEnabled = true             // Auto-sync in background

go build -o bin/litestream-test ./cmd/litestream-test

litestream-test populate -db <path> [options]

litestream-test populate -db /tmp/test.db -target-size 1GB
litestream-test populate -db /tmp/test.db -target-size 50MB -batch-size 10000
litestream-test populate -db /tmp/test.db -target-size 1.5GB -page-size 4096

litestream-test load -db <path> [options]

litestream-test load -db /tmp/test.db -write-rate 50 -duration 5m
litestream-test load -db /tmp/test.db -write-rate 100 -duration 2h -pattern wave
litestream-test load -db /tmp/test.db -write-rate 200 -duration 8h -workers 4 -pattern burst

litestream-test shrink -db <path> [options]

litestream-test validate [options]

litestream-test validate -source-db /tmp/test.db -replica-url file:///tmp/replica
litestream-test validate -source-db /tmp/test.db -replica-url s3://bucket/path -check-type full
litestream-test validate -source-db /tmp/test.db -config /tmp/litestream.yml -ltx-continuity

litestream-test version

litestream-test populate -db /tmp/test.db -target-size 100MB

litestream replicate /tmp/test.db file:///tmp/replica &
LITESTREAM_PID=$!

litestream-test load -db /tmp/test.db -duration 5m -write-rate 50

kill $LITESTREAM_PID
wait

litestream-test validate -source-db /tmp/test.db -replica-url file:///tmp/replica

litestream-test populate -db /tmp/test.db -target-size 100MB

litestream replicate -config litestream.yml &

litestream-test load -db /tmp/test.db \
  -duration 8h \
  -write-rate 100 \
  -pattern wave \
  -workers 4

litestream-test populate -db /tmp/test.db \
  -target-size 1.5GB \
  -page-size 4096 \
  -batch-size 10000

litestream replicate /tmp/test.db s3://bucket/path &

litestream-test validate -source-db /tmp/test.db \
  -replica-url s3://bucket/path \
  -check-type full

go build -o bin/litestream-test ./cmd/litestream-test

go test ./cmd/litestream-test/...

pip install litestream-vfs

import sqlite3
import litestream_vfs

conn = sqlite3.connect(":memory:")
litestream_vfs.load(conn)

path = litestream_vfs.loadable_path()

gem install litestream-vfs

gem "litestream-vfs"

require "litestream_vfs"

db = SQLite3::Database.new(":memory:")
LitestreamVfs.load(db)

path = LitestreamVfs.loadable_path

# Test for 60 seconds
./test-cpu-usage.sh 60

# Longer test (5 minutes)
./test-cpu-usage.sh 300

go build -o bin/litestream ./cmd/litestream
go build -o bin/litestream-test ./cmd/litestream-test

go test -v -tags=integration -timeout=30m ./tests/integration/... \
  -run="TestFreshStart|TestDatabaseIntegrity|TestRapidCheckpoints"

go test -v -tags=integration -timeout=1h ./tests/integration/...

go test -v -tags="integration,long" -timeout=10h ./tests/integration/... \
  -run="TestOvernight|Test1GBBoundary"

go test -v -tags="integration,soak" \
  -run=TestComprehensiveSoak -timeout=3h ./tests/integration

SOAK_KEEP_TEMP=1 SOAK_DEBUG=1 go test -v -tags="integration,soak" \
  -run=TestComprehensiveSoak -test.short -timeout=1h ./tests/integration

SOAK_AUTO_PURGE=yes SOAK_KEEP_TEMP=1 go test -v -tags="integration,soak,docker" \
  -run=TestMinIOSoak -test.short -timeout=20m ./tests/integration

export AWS_ACCESS_KEY_ID=...
export AWS_SECRET_ACCESS_KEY=...
export S3_BUCKET=your-bucket
export AWS_REGION=us-east-1

go test -v -tags="integration,soak,aws" \
  -run=TestOvernightS3Soak -timeout=10h ./tests/integration

# Fresh start test
go test -v -tags=integration ./tests/integration/... -run=TestFreshStart

# Rapid checkpoints test
go test -v -tags=integration ./tests/integration/... -run=TestRapidCheckpoints

# 1GB boundary test
go test -v -tags=integration ./tests/integration/... -run=Test1GBBoundary

go test -v -tags=integration -short ./tests/integration/...

- Quick integration tests (TestFreshStart, TestDatabaseIntegrity, TestRapidCheckpoints)
- Timeout: 30 minutes

workflow_dispatch → test_type: quick

workflow_dispatch → test_type: all

workflow_dispatch → test_type: long

/tmp/<test-temp-dir>/
├── <name>.db          # Test database
├── <name>.db-wal      # WAL file
├── <name>.db-shm      # Shared memory
├── replica/           # Replica directory
│   └── ltx/0/        # LTX files
├── litestream.log     # Litestream output
└── *-restored.db      # Restored databases

log, err := db.GetLitestreamLog()
fmt.Println(log)

errors, err := db.CheckForErrors()
for _, e := range errors {
    t.Logf("Error: %s", e)
}

fileCount, _ := db.GetReplicaFileCount()
t.Logf("LTX files: %d", fileCount)

size, _ := db.GetDatabaseSize()
t.Logf("DB size: %.2f MB", float64(size)/(1024*1024))

go test -v -tags="integration,soak" -timeout=3h -run=TestComprehensiveSoak ./tests/integration/

# Ensure Docker is running
go test -v -tags="integration,soak,docker" -timeout=3h -run=TestMinIOSoak ./tests/integration/

export AWS_ACCESS_KEY_ID=your_key
export AWS_SECRET_ACCESS_KEY=your_secret
export S3_BUCKET=your-test-bucket
export AWS_REGION=us-east-1

go test -v -tags="integration,soak,aws" -timeout=10h -run=TestOvernightS3Soak ./tests/integration/

go test -v -tags="integration,soak,docker,aws" -timeout=15h ./tests/integration/

# Run comprehensive test for 30 minutes instead of 2 hours
go test -v -tags="integration,soak" -timeout=1h -run=TestComprehensiveSoak ./tests/integration/ -test.short

# Watch test progress in real-time
go test -v -tags="integration,soak" -run=TestComprehensiveSoak ./tests/integration/ 2>&1 | tee soak-test.log

go build -o bin/litestream ./cmd/litestream
go build -o bin/litestream-test ./cmd/litestream-test

./cmd/litestream-test/scripts/verify-test-setup.sh

./cmd/litestream-test/scripts/verify-test-setup.sh

./cmd/litestream-test/scripts/reproduce-critical-bug.sh

./cmd/litestream-test/scripts/test-754-s3-scenarios.sh

./cmd/litestream-test/scripts/test-754-restore-focus.sh

./cmd/litestream-test/scripts/test-simple-754-reproduction.sh

./cmd/litestream-test/scripts/test-v0.5-flag-reproduction.sh

./cmd/litestream-test/scripts/test-v0.5-restart-scenarios.sh

./cmd/litestream-test/scripts/test-format-isolation.sh

./cmd/litestream-test/scripts/test-quick-format-check.sh

./cmd/litestream-test/scripts/test-upgrade-v0.3-to-v0.5.sh

./cmd/litestream-test/scripts/test-upgrade-large-db.sh

./cmd/litestream-test/scripts/test-massive-upgrade.sh

export LITESTREAM_S3_ACCESS_POINT_ARN='arn:aws:s3:us-east-2:123456789012:accesspoint/my-access-point'
./cmd/litestream-test/scripts/test-s3-access-point.sh

./cmd/litestream-test/scripts/test-s3-retention-cleanup.sh

./cmd/litestream-test/scripts/test-s3-retention-small-db.sh

./cmd/litestream-test/scripts/test-s3-retention-large-db.sh

./cmd/litestream-test/scripts/test-s3-retention-comprehensive.sh

./cmd/litestream-test/scripts/test-s3-retention-comprehensive.sh --small-only
./cmd/litestream-test/scripts/test-s3-retention-comprehensive.sh --large-only
./cmd/litestream-test/scripts/test-s3-retention-comprehensive.sh --no-cleanup

./cmd/litestream-test/scripts/test-fresh-start.sh

./cmd/litestream-test/scripts/verify-test-setup.sh
./cmd/litestream-test/scripts/test-rapid-checkpoints.sh

for script in test-fresh-start.sh test-rapid-checkpoints.sh test-database-integrity.sh; do
    echo "Running $script..."
    ./cmd/litestream-test/scripts/$script
    echo ""
done

./cmd/litestream-test/scripts/test-s3-retention-small-db.sh

tail -f /tmp/fresh-test.log

tail -f /tmp/checkpoint-cycle.log

grep -i error /tmp/*.log

npm install litestream-vfs

const { getLoadablePath } = require("litestream-vfs");
const path = getLoadablePath();
// Use `path` with better-sqlite3 or other SQLite bindings

graph TB
    subgraph "Application Layer"
        CLI[CLI Commands<br/>cmd/litestream/]
        Config[Configuration<br/>config.go]
    end

    subgraph "Core Layer"
        Store[Store Manager<br/>store.go]
        DB[Database Manager<br/>db.go]
        Replica[Replica Manager<br/>replica.go]
    end

    subgraph "Storage Abstraction"
        RC[ReplicaClient Interface<br/>replica_client.go]
    end

    subgraph "Storage Implementations"
        S3[s3/replica_client.go]
        GCS[gs/replica_client.go]
        ABS[abs/replica_client.go]
        OSS[oss/replica_client.go]
        File[file/replica_client.go]
        SFTP[sftp/replica_client.go]
        NATS[nats/replica_client.go]
    end

    subgraph "External"
        SQLite[SQLite Database]
        Cloud[Cloud Storage]
    end

    CLI --> Store
    Store --> DB
    DB --> Replica
    Replica --> RC
    RC --> S3
    RC --> GCS
    RC --> ABS
    RC --> OSS
    RC --> File
    RC --> SFTP
    RC --> NATS
    DB <--> SQLite
    S3 --> Cloud
    GCS --> Cloud
    ABS --> Cloud
    OSS --> Cloud

type DB struct {
    // Core fields
    path     string      // Database file path
    metaPath string      // Metadata directory path
    db       *sql.DB     // SQLite connection
    f        *os.File    // Long-running file descriptor
    rtx      *sql.Tx     // Long-running read transaction
    pageSize int         // Database page size

    // Synchronization
    mu       sync.RWMutex   // Protects struct fields
    chkMu    sync.RWMutex   // Checkpoint lock
    notify   chan struct{}  // WAL change notifications

    // Lifecycle
    ctx    context.Context
    cancel func()
    wg     sync.WaitGroup

    // Configuration
    MinCheckpointPageN int              // Min pages for passive checkpoint
    TruncatePageN      int              // Pages before emergency truncate checkpoint
    CheckpointInterval time.Duration    // Time-based passive checkpoint interval
    MonitorInterval    time.Duration    // WAL monitoring frequency
    // Note: MaxCheckpointPageN removed (RESTART mode disabled due to #724)

    // Metrics
    dbSizeGauge        prometheus.Gauge
    walSizeGauge       prometheus.Gauge
    txIDGauge          prometheus.Gauge
}

// Lifecycle
func (db *DB) Open() error
func (db *DB) Close(ctx context.Context) error

// Monitoring
func (db *DB) monitor()              // Background WAL monitoring
func (db *DB) checkWAL() (bool, error)  // Check for WAL changes

// Checkpointing
func (db *DB) Checkpoint(mode string) error
func (db *DB) autoCheckpoint() error

// Replication
func (db *DB) WALReader(pgno uint32) (io.ReadCloser, error)
func (db *DB) Sync(ctx context.Context) error

// Compaction
func (db *DB) Compact(ctx context.Context, destLevel int) (*ltx.FileInfo, error)

type Replica struct {
    db *DB                    // Parent database
    Client ReplicaClient      // Storage backend client

    mu  sync.RWMutex
    pos ltx.Pos              // Current replication position

    // Configuration
    SyncInterval time.Duration
    MonitorEnabled bool

    // Lifecycle
    cancel func()
    wg     sync.WaitGroup
}

type Pos struct {
    TXID     TXID      // Transaction ID
    PageNo   uint32    // Page number within transaction
    Checksum uint64    // Running checksum
}

type Store struct {
    mu     sync.Mutex
    dbs    []*DB
    levels CompactionLevels

    // Configuration
    SnapshotInterval           time.Duration
    SnapshotRetention          time.Duration
    L0Retention                time.Duration
    L0RetentionCheckInterval   time.Duration
    CompactionMonitorEnabled bool

    // Lifecycle
    ctx    context.Context
    cancel func()
    wg     sync.WaitGroup
}

type SocketConfig struct {
    Enabled     bool   `yaml:"enabled"`     // Default: false
    Path        string `yaml:"path"`        // Default: "/var/run/litestream.sock"
    Permissions uint32 `yaml:"permissions"` // Default: 0600
}

type RegisterDatabaseRequest struct {
    Path       string `json:"path"`
    ReplicaURL string `json:"replica_url"`
}

type UnregisterDatabaseRequest struct {
    Path    string `json:"path"`
    Timeout int    `json:"timeout,omitempty"` // Seconds
}

type TXIDResponse struct {
    TXID uint64 `json:"txid"`
}

type Leaser interface {
    Type() string
    AcquireLease(ctx context.Context) (*Lease, error)
    RenewLease(ctx context.Context, lease *Lease) (*Lease, error)
    ReleaseLease(ctx context.Context, lease *Lease) error
}

type Lease struct {
    Generation int64     `json:"generation"`
    ExpiresAt  time.Time `json:"expires_at"`
    Owner      string    `json:"owner,omitempty"`
    ETag       string    `json:"-"`
}

type SyncStatus struct {
    LocalTXID  ltx.TXID // Local transaction ID
    RemoteTXID ltx.TXID // Remote transaction ID
    InSync     bool     // true if LocalTXID > 0 and equal to RemoteTXID
}

func (db *DB) SyncStatus(ctx context.Context) (SyncStatus, error)
func (db *DB) SyncAndWait(ctx context.Context) error
func (db *DB) EnsureExists(ctx context.Context) error

+------------------+
|     Header       |  Fixed size header with metadata
+------------------+
|                  |
|   Page Frames    |  Variable number of page frames
|                  |
+------------------+
|   Page Index     |  Index for efficient page lookup
+------------------+
|     Trailer      |  Metadata and checksums
+------------------+

type Header struct {
    Magic       [4]byte  // "LTX\x00"
    Version     uint32   // Format version
    PageSize    uint32   // Database page size
    MinTXID     TXID     // Starting transaction ID
    MaxTXID     TXID     // Ending transaction ID
    Timestamp   int64    // Creation timestamp
    Checksum    uint64   // Header checksum
}

type PageFrame struct {
    Header PageHeader
    Data   []byte     // Page data (pageSize bytes)
}

type PageHeader struct {
    PageNo   uint32    // Page number in database
    Size     uint32    // Size of page data
    Checksum uint64    // Page checksum
}

type PageIndexElem struct {
    PageNo uint32    // Page number
    Offset int64     // Offset in file
    Size   uint32    // Size of page frame
}

type Trailer struct {
    PageIndexOffset int64   // Offset to page index
    PageIndexSize   int64   // Size of page index
    PageCount       uint32  // Total pages in file
    Checksum        uint64  // Full file checksum
}

func (db *DB) monitor() {
    ticker := time.NewTicker(db.MonitorInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            // Check WAL for changes
            changed, err := db.checkWAL()
            if err != nil {
                slog.Error("wal check failed", "error", err)
                continue
            }

            if changed {
                // Notify replicas of changes
                db.notifyReplicas()

                // Check if checkpoint needed
                if db.shouldCheckpoint() {
                    db.autoCheckpoint()
                }
            }

        case <-db.ctx.Done():
            return
        }
    }
}

func (db *DB) checkWAL() (bool, error) {
    // Get current WAL state
    walInfo, err := db.walInfo()
    if err != nil {
        return false, err
    }

    // Compare with previous state
    db.mu.Lock()
    changed := walInfo.Size != db.prevWALSize ||
               walInfo.Checksum != db.prevWALChecksum
    db.prevWALSize = walInfo.Size
    db.prevWALChecksum = walInfo.Checksum
    db.mu.Unlock()

    return changed, nil
}

type CompactionLevel struct {
    Level    int           // Level number (0 = raw, 1+ = compacted)
    Interval time.Duration // How often to compact from previous level
}

// Default configuration
var DefaultCompactionLevels = CompactionLevels{
    {Level: 0, Interval: 0},          // Raw LTX files
    {Level: 1, Interval: 1 * Hour},   // Hourly compaction
    {Level: 2, Interval: 24 * Hour},  // Daily compaction
}

func (db *DB) initReadTx() error {
    // Start read transaction
    tx, err := db.db.BeginTx(context.Background(), &sql.TxOptions{
        ReadOnly: true,
    })
    if err != nil {
        return err
    }

    // Execute dummy query to start transaction
    var dummy string
    err = tx.QueryRow("SELECT ''").Scan(&dummy)
    if err != nil {
        tx.Rollback()
        return err
    }

    db.rtx = tx
    return nil
}

func (db *DB) Checkpoint(mode string) error {
    // Acquire checkpoint lock
    db.chkMu.Lock()
    defer db.chkMu.Unlock()

    // Close read transaction temporarily
    if db.rtx != nil {
        db.rtx.Rollback()
        db.rtx = nil
    }

    // Perform checkpoint
    _, _, err := db.db.Exec(fmt.Sprintf("PRAGMA wal_checkpoint(%s)", mode))
    if err != nil {
        return err
    }

    // Restart read transaction
    return db.initReadTx()
}

// DB struct mutexes
type DB struct {
    mu     sync.RWMutex  // Protects struct fields
    chkMu  sync.RWMutex  // Checkpoint coordination
}

// Replica struct mutexes
type Replica struct {
    mu  sync.RWMutex    // Protects position
    muf sync.Mutex      // File descriptor lock
}

// Store struct mutex
type Store struct {
    mu sync.Mutex       // Protects database list
}

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(50) // Max 50 concurrent DB opens
for _, db := range s.dbs {
    db := db
    g.Go(func() error { return db.Open() })
}

// Start background task
func (db *DB) Start() {
    db.wg.Add(1)
    go func() {
        defer db.wg.Done()
        db.monitor()
    }()
}

// Stop with timeout
func (db *DB) Close(ctx context.Context) error {
    // Signal shutdown
    db.cancel()

    // Wait for goroutines with timeout
    done := make(chan struct{})
    go func() {
        db.wg.Wait()
        close(done)
    }()

    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

stateDiagram-v2
    [*] --> Closed
    Closed --> Opening: Open()
    Opening --> Open: Success
    Opening --> Closed: Error
    Open --> Monitoring: Start()
    Monitoring --> Syncing: Changes Detected
    Syncing --> Monitoring: Sync Complete
    Monitoring --> Checkpointing: Threshold Reached
    Checkpointing --> Monitoring: Checkpoint Complete
    Monitoring --> Closing: Close()
    Closing --> Closed: Cleanup Complete

stateDiagram-v2
    [*] --> Idle
    Idle --> Starting: Start()
    Starting --> Monitoring: Success
    Starting --> Idle: Error
    Monitoring --> Syncing: Timer/Changes
    Syncing --> Uploading: Have Changes
    Uploading --> Monitoring: Success
    Uploading --> Error: Failed
    Error --> Monitoring: Retry
    Monitoring --> Stopping: Stop()
    Stopping --> Idle: Cleanup

type Pos struct {
    TXID     TXID      // Current transaction ID
    PageNo   uint32    // Current page number
    Checksum uint64    // Running checksum for validation
}

// Update position atomically
func (r *Replica) SetPos(pos ltx.Pos) {
    r.mu.Lock()  // MUST use Lock, not RLock!
    defer r.mu.Unlock()
    r.pos = pos
}

// Read position safely
func (r *Replica) Pos() ltx.Pos {
    r.mu.RLock()
    defer r.mu.RUnlock()
    return r.pos
}

sequenceDiagram
    participant Main
    participant Store
    participant DB
    participant Replica
    participant Monitor

    Main->>Store: NewStore(config)
    Store->>Store: Validate config

    Main->>Store: Open()
    loop For each database
        Store->>DB: NewDB(path)
        Store->>DB: Open()
        DB->>DB: Open SQLite connection
        DB->>DB: Read page size
        DB->>DB: Init metadata
        DB->>DB: Start read transaction

        loop For each replica
            DB->>Replica: NewReplica()
            DB->>Replica: Start()
            Replica->>Monitor: Start monitoring
        end
    end

    Store->>Store: Start compaction monitors
    Store-->>Main: Ready

// Bottom-up error propagation
ReplicaClient.WriteLTXFile() error
    ↓
Replica.Sync() error
    ↓
DB.Sync() error
    ↓
Store.monitorDB() // Logs error, continues

func (r *Replica) syncWithRetry(ctx context.Context) error {
    backoff := time.Second
    maxBackoff := time.Minute

    for attempt := 0; ; attempt++ {
        err := r.Sync(ctx)
        if err == nil {
            return nil
        }

        // Check if error is retryable
        if !isRetryable(err) {
            return err
        }

        // Check context
        if ctx.Err() != nil {
            return ctx.Err()
        }

        // Exponential backoff
        time.Sleep(backoff)
        backoff *= 2
        if backoff > maxBackoff {
            backoff = maxBackoff
        }
    }
}

// Database metrics
db_size_bytes           // Current database size
wal_size_bytes         // Current WAL size
total_wal_bytes        // Total bytes written to WAL
checkpoint_count       // Number of checkpoints
sync_count            // Number of syncs
sync_error_count      // Number of sync errors

// Replica metrics
replica_lag_seconds    // Replication lag
replica_position      // Current replication position

// Validation metrics have db and level labels.
litestream_validation_checks_total                   // Checks by result: success, invalid, or error.
litestream_last_validation_success                   // Last check: 1 passed, 0 failed or not yet complete.
litestream_last_validation_success_timestamp_seconds // Unix time of last success, or 0 until a check passes.

func (db *DB) HealthCheck() error {
    // Check database connection
    if err := db.db.Ping(); err != nil {
        return fmt.Errorf("database ping failed: %w", err)
    }

    // Check replication lag
    for _, r := range db.replicas {
        lag := time.Since(r.LastSync())
        if lag > MaxAcceptableLag {
            return fmt.Errorf("replica %s lag too high: %v", r.Name(), lag)
        }
    }

    return nil
}

[
  {
    "path": "/var/lib/app.db",
    "replica": "s3"
  }
]

{
  "version": "v0.5.0",
  "pid": 12345,
  "uptime_seconds": 300,
  "started_at": "2026-04-24T12:00:00Z",
  "database_count": 2
}

{
  "databases": [
    {
      "path": "/var/lib/app.db",
      "status": "replicating",
      "last_sync_at": "2026-04-24T12:00:00Z"
    }
  ]
}

[
  {
    "level": 0,
    "min_txid": "0000000000000001",
    "max_txid": "0000000000000004",
    "size": 8192,
    "timestamp": "2026-04-24T12:00:00Z"
  }
]

{
  "db_path": "/var/lib/app.db",
  "replica": "file",
  "txid": "0000000000000004",
  "duration_ms": 125,
  "integrity_check": "quick"
}

{
  "source": "file:///backups/app.db",
  "target_path": "/var/lib/app.db",
  "replica": "file",
  "min_txid": "0000000000000001",
  "max_txid": "0000000000000004",
  "files": [
    {
      "level": 9,
      "name": "0000000000000001-0000000000000004.ltx",
      "min_txid": "0000000000000001",
      "max_txid": "0000000000000004",
      "size": 8192,
      "timestamp": "2026-04-24T12:00:00Z"
    }
  ]
}

[
  {
    "database": "/var/lib/app.db",
    "status": "ok",
    "local_txid": "0000000000000004",
    "wal_size": "32 kB"
  }
]

### How to initialize DB

```go
func (db *DB) init() {
    db.mu.Lock()
    defer db.mu.Unlock()
    // ... 50 lines of code copied from db.go
}
\```

### DB Initialization Pattern

**Principle**: Database initialization must complete before replication starts.

**Pattern**:

1. Acquire exclusive lock (`mu.Lock()`)
2. Verify database state consistency
3. Initialize monitoring subsystems
4. Set up replication coordination

**Critical**: Use `Lock()` not `RLock()` as initialization modifies state.

**Reference Implementation**: See `DB.init()` in db.go:150-230

# Find all code examples in documentation
rg "^```(go|golang)" docs/ CLAUDE.md AGENTS.md .claude/

# Find file:line references
rg "\.go:\d+" docs/ CLAUDE.md AGENTS.md .claude/

# Find interface definitions
rg "type .* interface" docs/ CLAUDE.md AGENTS.md .claude/

# Lint all docs
markdownlint docs/ CLAUDE.md AGENTS.md .claude/

# Auto-fix issues
markdownlint --fix docs/ CLAUDE.md AGENTS.md .claude/

# List all go files mentioned in docs
rg -o "[a-z_]+\.go:\d+" docs/ CLAUDE.md AGENTS.md | sort -u

# Verify they exist and line numbers are reasonable

graph LR
    WAL[SQLite WAL] -->|Convert| LTX[LTX File]
    LTX -->|Upload| Storage[Cloud Storage]
    Storage -->|Download| Restore[Restored DB]

┌─────────────────────┐
│      Header         │ Fixed size (varies by version)
├─────────────────────┤
│                     │
│    Page Frames      │ Variable number of pages
│                     │
├─────────────────────┤
│    Page Index       │ Binary search tree
├─────────────────────┤
│      Trailer        │ Fixed size metadata
└─────────────────────┘

FileSize = HeaderSize +
           (PageCount * (PageHeaderSize + PageSize)) +
           PageIndexSize +
           TrailerSize

// From github.com/superfly/ltx
type Header struct {
    Version          int      // Derived from the magic string ("LTX1")
    Flags            uint32   // Reserved flag bits
    PageSize         uint32   // Database page size
    Commit           uint32   // Page count after applying file
    MinTXID          TXID
    MaxTXID          TXID
    Timestamp        int64    // Milliseconds since Unix epoch
    PreApplyChecksum Checksum // Database checksum before apply
    WALOffset        int64    // Offset within source WAL (0 for snapshots)
    WALSize          int64    // WAL byte length (0 for snapshots)
    WALSalt1         uint32
    WALSalt2         uint32
    NodeID           uint64
}

const HeaderFlagNoChecksum = uint32(1 << 1)

Offset  Size  Field
0       4     Magic ("LTX1")
4       4     Flags
8       4     PageSize
12      4     Commit
16      8     MinTXID
24      8     MaxTXID
32      8     Timestamp
40      8     PreApplyChecksum
48      8     WALOffset
56      8     WALSize
64      4     WALSalt1
68      4     WALSalt2
72      8     NodeID
80     20     Reserved (zeros)
Total: 100 bytes

type PageFrame struct {
    Header PageHeader
    Data   []byte  // Size = PageSize from LTX header
}

type PageHeader struct {
    Pgno uint32  // Database page number (1-based)
}

Offset  Size     Field
0       4        Page Number (Pgno)
4       PageSize Page Data

type PageIndexElem struct {
    Level   int
    MinTXID TXID
    MaxTXID TXID
    Offset  int64 // Byte offset of encoded payload
    Size    int64 // Bytes occupied by encoded payload
}

Rather than parsing raw bytes, call `ltx.DecodePageIndex` which returns a
map of page number to `ltx.PageIndexElem` for you.

// Finding a page using the index
func findPage(index []PageIndexElem, targetPageNo uint32) (offset int64, found bool) {
    // Binary search
    idx := sort.Search(len(index), func(i int) bool {
        return index[i].PageNo >= targetPageNo
    })

    if idx < len(index) && index[idx].PageNo == targetPageNo {
        return index[idx].Offset, true
    }
    return 0, false
}

type Trailer struct {
    PostApplyChecksum Checksum // Database checksum after apply
    FileChecksum      Checksum // CRC-64 checksum of entire file
}

Offset  Size  Field
0       8     PostApplyChecksum
8       8     FileChecksum
Total: 16 bytes

func readTrailer(f *os.File) (*Trailer, error) {
    // Seek to trailer position
    _, err := f.Seek(-TrailerSize, io.SeekEnd)
    if err != nil {
        return nil, err
    }

    var trailer Trailer
    err = binary.Read(f, binary.BigEndian, &trailer)
    return &trailer, err
}

Format: MMMMMMMMMMMMMMMM-NNNNNNNNNNNNNNNN.ltx
Where:
  M = MinTXID (16 hex digits, zero-padded)
  N = MaxTXID (16 hex digits, zero-padded)

Examples:
  0000000000000001-0000000000000064.ltx  (TXID 1-100)
  0000000000000065-00000000000000c8.ltx  (TXID 101-200)

// From github.com/superfly/ltx
func ParseFilename(name string) (minTXID, maxTXID TXID, err error) {
    // Remove extension
    name = strings.TrimSuffix(name, ".ltx")

    // Split on hyphen
    parts := strings.Split(name, "-")
    if len(parts) != 2 {
        return 0, 0, errors.New("invalid format")
    }

    // Parse hex values
    min, err := strconv.ParseUint(parts[0], 16, 64)
    max, err := strconv.ParseUint(parts[1], 16, 64)

    return TXID(min), TXID(max), nil
}

func FormatFilename(minTXID, maxTXID TXID) string {
    return fmt.Sprintf("%016x-%016x.ltx", minTXID, maxTXID)
}

import "hash/crc64"

var crcTable = crc64.MakeTable(crc64.ECMA)

func calculateChecksum(data []byte) uint64 {
    return crc64.Checksum(data, crcTable)
}

// Cumulative checksum for multiple pages
func cumulativeChecksum(pages [][]byte) uint64 {
    h := crc64.New(crcTable)
    for _, page := range pages {
        h.Write(page)
    }
    return h.Sum64()
}

func verifyPage(header PageHeader, data []byte) error {
    if header.Checksum == 0 {
        return nil // Checksums disabled
    }

    calculated := calculateChecksum(data)
    if calculated != header.Checksum {
        return fmt.Errorf("checksum mismatch: expected %x, got %x",
            header.Checksum, calculated)
    }
    return nil
}

Level 0: Raw files (no compaction)
         /ltx/0000/0000000000000001-0000000000000064.ltx
         /ltx/0000/0000000000000065-00000000000000c8.ltx

Level 1: Hourly compaction
         /ltx/0001/0000000000000001-0000000000000fff.ltx

Level 2: Daily compaction
         /ltx/0002/0000000000000001-000000000000ffff.ltx

Snapshots: Full database state
          /snapshots/20240101120000.ltx

func compactLTXFiles(files []*LTXFile) (*LTXFile, error) {
    // Create page map (newer overwrites older)
    pageMap := make(map[uint32]Page)

    for _, file := range files {
        for _, page := range file.Pages {
            pageMap[page.Number] = page
        }
    }

    // Create new LTX with merged pages
    merged := &LTXFile{
        MinTXID: files[0].MinTXID,
        MaxTXID: files[len(files)-1].MaxTXID,
    }

    // Add pages in order (skip lock page)
    for pgno := uint32(1); pgno <= maxPgno; pgno++ {
        if pgno == LockPageNumber(pageSize) {
            continue // Skip 1GB lock page
        }
        if page, ok := pageMap[pgno]; ok {
            merged.Pages = append(merged.Pages, page)
        }
    }

    return merged, nil
}

func ReadLTXFile(path string) (*LTXFile, error) {
    f, err := os.Open(path)
    if err != nil {
        return nil, err
    }
    defer f.Close()

    dec := ltx.NewDecoder(f)

    // Read and verify header
    header, err := dec.Header()
    if err != nil {
        return nil, err
    }

    // Read all pages
    var pages []Page
    for {
        var pageHeader ltx.PageHeader
        pageData := make([]byte, header.PageSize)

        err := dec.DecodePage(&pageHeader, pageData)
        if err == io.EOF {
            break
        }
        if err != nil {
            return nil, err
        }

        pages = append(pages, Page{
            Number: pageHeader.PageNo,
            Data:   pageData,
        })
    }

    return &LTXFile{
        Header: header,
        Pages:  pages,
    }, nil
}

func ReadPage(path string, pageNo uint32) ([]byte, error) {
    f, err := os.Open(path)
    if err != nil {
        return nil, err
    }
    defer f.Close()

    // Read trailer to find index
    trailer, err := readTrailer(f)
    if err != nil {
        return nil, err
    }

    // Read page index
    f.Seek(trailer.PageIndexOffset, io.SeekStart)
    indexData := make([]byte, trailer.PageIndexSize)
    f.Read(indexData)

    index := parsePageIndex(indexData)

    // Find page in index
    offset, found := findPage(index, pageNo)
    if !found {
        return nil, errors.New("page not found")
    }

    // Read page at offset
    f.Seek(offset, io.SeekStart)

    var pageHeader PageHeader
    binary.Read(f, binary.BigEndian, &pageHeader)

    pageData := make([]byte, pageSize)
    f.Read(pageData)

    return pageData, nil
}

func WriteLTXFile(path string, pages []Page) error {
    f, err := os.Create(path)
    if err != nil {
        return err
    }
    defer f.Close()

    enc := ltx.NewEncoder(f)

    // Write header
    header := ltx.Header{
        Version:   ltx.Version,
        Flags:     0,
        PageSize:  4096,
        PageCount: uint32(len(pages)),
        MinTXID:   minTXID,
        MaxTXID:   maxTXID,
    }

    if err := enc.EncodeHeader(header); err != nil {
        return err
    }

    // Write pages and build index
    var index []PageIndexElem
    for _, page := range pages {
        offset := enc.Offset()

        // Skip lock page
        if page.Number == LockPageNumber(header.PageSize) {
            continue
        }

        pageHeader := ltx.PageHeader{
            PageNo:   page.Number,
            Checksum: calculateChecksum(page.Data),
        }

        if err := enc.EncodePage(pageHeader, page.Data); err != nil {
            return err
        }

        index = append(index, PageIndexElem{
            PageNo: page.Number,
            Offset: offset,
        })
    }

    // Write page index
    if err := enc.EncodePageIndex(index); err != nil {
        return err
    }

    // Write trailer
    if err := enc.EncodeTrailer(); err != nil {
        return err
    }

    return enc.Close()
}

sequenceDiagram
    participant SQLite
    participant WAL
    participant Litestream
    participant LTX

    SQLite->>WAL: Write transaction
    WAL->>WAL: Append frames

    Litestream->>WAL: Monitor changes
    WAL-->>Litestream: Read frames

    Litestream->>Litestream: Convert frames
    Note over Litestream: - Skip lock page- Add checksums- Build index

    Litestream->>LTX: Write LTX file
    LTX->>Storage: Upload

type TXID uint64

// TXID represents a logical transaction boundary
// Not directly from SQLite, but derived from:
// 1. WAL checkpoint sequence
// 2. Frame count
// 3. Logical grouping of changes

func (db *DB) nextTXID() TXID {
    // Increment from last known TXID
    return db.lastTXID + 1
}

const PENDING_BYTE = 0x40000000

func shouldSkipPage(pageNo uint32, pageSize int) bool {
    lockPage := uint32(PENDING_BYTE/pageSize) + 1
    return pageNo == lockPage
}

// Keep earliest CreatedAt from source files
func compactWithTimestamp(files []*FileInfo) *FileInfo {
    earliest := files[0].CreatedAt
    for _, f := range files[1:] {
        if f.CreatedAt.Before(earliest) {
            earliest = f.CreatedAt
        }
    }

    return &FileInfo{
        CreatedAt: earliest, // Preserve for point-in-time recovery
    }
}

func safeReadLTX(path string) (*LTXFile, error) {
    file, err := ReadLTXFile(path)
    if err != nil {
        return nil, err
    }

    // Verify all checksums
    for _, page := range file.Pages {
        if err := verifyPage(page); err != nil {
            return nil, fmt.Errorf("corrupted page %d: %w",
                page.Number, err)
        }
    }

    return file, nil
}

// For eventually consistent storage
func readWithRetry(client ReplicaClient, info *FileInfo) ([]byte, error) {
    for attempts := 0; attempts < 5; attempts++ {
        data, err := client.OpenLTXFile(...)
        if err == nil {
            // Verify we got complete file
            if int64(len(data)) == info.Size {
                return data, nil
            }
        }

        time.Sleep(time.Second * time.Duration(attempts+1))
    }

    return nil, errors.New("incomplete file after retries")
}

litestream ltx /path/to/db.sqlite
litestream ltx s3://bucket/db

f, err := os.Open("0000000000000001-0000000000000064.ltx")
if err != nil {
    log.Fatal(err)
}
defer f.Close()

dec := ltx.NewDecoder(f)
if err := dec.DecodeHeader(); err != nil {
    log.Fatal(err)
}
for {
    var hdr ltx.PageHeader
    data := make([]byte, dec.Header().PageSize)
    if err := dec.DecodePage(&hdr, data); err == io.EOF {
        break
    } else if err != nil {
        log.Fatal(err)
    }
    // Inspect hdr.Pgno or data here.
}
if err := dec.Close(); err != nil {
    log.Fatal(err)
}
fmt.Println("post-apply checksum:", dec.Trailer().PostApplyChecksum)

DB Layer (db.go)          → Database state, restoration, monitoring
Replica Layer (replica.go) → Replication mechanics only
Storage Layer             → ReplicaClient implementations

// CORRECT - DB layer handles database state
func (db *DB) init() error {
    // DB layer handles database state
    if db.needsRestore() {
        if err := db.restore(); err != nil {
            return err
        }
    }
    // Then start replica for replication only
    return db.replica.Start()
}

func (r *Replica) Start() error {
    // Replica focuses only on replication
    return r.startSync()
}

// WRONG - Replica should only handle replication concerns
func (r *Replica) Start() error {
    // DON'T check database state here
    if needsRestore() {  // Wrong layer!
        restoreDatabase()  // Wrong layer!
    }
    // Replica should focus only on replication mechanics
}

// CORRECT - Atomic file write pattern
func writeFileAtomic(path string, data []byte) error {
    // Create temp file in same directory (for atomic rename)
    dir := filepath.Dir(path)
    tmpFile, err := os.CreateTemp(dir, ".tmp-*")
    if err != nil {
        return fmt.Errorf("create temp file: %w", err)
    }
    tmpPath := tmpFile.Name()

    // Clean up temp file on error
    defer func() {
        if tmpFile != nil {
            tmpFile.Close()
            os.Remove(tmpPath)
        }
    }()

    // Write data to temp file
    if _, err := tmpFile.Write(data); err != nil {
        return fmt.Errorf("write temp file: %w", err)
    }

    // Sync to ensure data is on disk
    if err := tmpFile.Sync(); err != nil {
        return fmt.Errorf("sync temp file: %w", err)
    }

    // Close before rename
    if err := tmpFile.Close(); err != nil {
        return fmt.Errorf("close temp file: %w", err)
    }
    tmpFile = nil // Prevent defer cleanup

    // Atomic rename (on same filesystem)
    if err := os.Rename(tmpPath, path); err != nil {
        os.Remove(tmpPath)
        return fmt.Errorf("rename to final path: %w", err)
    }

    return nil
}

// WRONG - Can leave partial files on failure
func writeFileDirect(path string, data []byte) error {
    return os.WriteFile(path, data, 0644)  // Not atomic!
}

// CORRECT - Return error for caller to handle
func (db *DB) validatePosition() error {
    dpos, err := db.Pos()
    if err != nil {
        return err
    }
    rpos := replica.Pos()
    if dpos.TXID < rpos.TXID {
        return fmt.Errorf("database position (%v) behind replica (%v)", dpos, rpos)
    }
    return nil
}

// WRONG - Silently continuing can cause data corruption
func (db *DB) validatePosition() {
    if dpos, _ := db.Pos(); dpos.TXID < replica.Pos().TXID {
        log.Printf("warning: position mismatch")  // Don't just log!
        // Continuing here is dangerous
    }
}

// WRONG - Continuing after error can corrupt state
func (db *DB) processFiles() {
    for _, file := range files {
        if err := processFile(file); err != nil {
            log.Printf("error: %v", err)  // Just logging!
            // Continuing to next file is dangerous
        }
    }
}

// CORRECT - Let caller decide how to handle errors
func (db *DB) processFiles() error {
    for _, file := range files {
        if err := processFile(file); err != nil {
            return fmt.Errorf("process file %s: %w", file, err)
        }
    }
    return nil
}

// CORRECT - Use Lock() for writes
r.mu.Lock()
defer r.mu.Unlock()
r.pos = pos

// WRONG - Race condition
r.mu.RLock()  // Should be Lock() for writes
defer r.mu.RUnlock()
r.pos = pos   // Writing with RLock!

// CORRECT - Check local first during compaction
// db.go:1280-1294 - ALWAYS read from local disk when available
f, err := os.Open(db.LTXPath(info.Level, info.MinTXID, info.MaxTXID))
if err == nil {
    // Use local file - it's complete and consistent
    return f, nil
}
// Only fall back to remote if local doesn't exist
return replica.Client.OpenLTXFile(...)

// WRONG - Can get partial/corrupt data from eventually consistent storage
f, err := client.OpenLTXFile(ctx, level, minTXID, maxTXID, 0, 0)

type LTXFileOpener interface {
    OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)
}

rc, _ := client.OpenLTXFile(ctx, level, min, max, 0, fileInfo.Size)
rr := internal.NewResumableReader(ctx, client, level, min, max, fileInfo.Size, rc, logger)
defer rr.Close()

rc, _ := client.OpenLTXFile(ctx, level, min, max, 0, 0)
// Risk: connection may drop during idle periods in multi-file restore
io.ReadFull(rc, buf) // unexpected EOF!

retention:
  enabled: false

store.SetRetentionEnabled(false) // Delegates deletion to cloud lifecycle policies

input := &s3.PutObjectInput{
    Bucket: aws.String(l.Bucket),
    Key:    aws.String(key),
    Body:   bytes.NewReader(data),
}
if etag == "" {
    input.IfNoneMatch = aws.String("*") // First acquire: only if key doesn't exist
} else {
    input.IfMatch = aws.String(etag)     // Expired takeover: only if ETag matches
}

_, err := l.s3.DeleteObject(ctx, &s3.DeleteObjectInput{
    Bucket:  aws.String(l.Bucket),
    Key:     aws.String(key),
    IfMatch: aws.String(lease.ETag), // Only delete if we still hold the lease
})

// CORRECT - Preserve temporal information
info, err := replica.Client.WriteLTXFile(ctx, level, minTXID, maxTXID, r)
if err != nil {
    return fmt.Errorf("write ltx: %w", err)
}
info.CreatedAt = oldestSourceFile.CreatedAt

// WRONG - Loses timestamp granularity for point-in-time restores
info := &ltx.FileInfo{
    CreatedAt: time.Now(), // Don't use current time during compaction
}

// WRONG - Database state logic in Replica layer
func (r *Replica) Start() error {
    if db.needsRestore() {  // Wrong layer for DB state!
        r.restoreDatabase()  // Replica shouldn't manage DB state!
    }
    return r.sync()
}

// WRONG - Don't reimplement what already exists
func customSnapshotTrigger() {
    // Complex custom logic to trigger snapshots
    // when db.verify() already does this!
}

// CORRECT - Use what's already there
func triggerSnapshot() error {
    return db.verify()  // Already handles snapshot logic correctly
}

// db.go:951-953 - Must skip lock page during replication
lockPgno := ltx.LockPgno(pageSize)
if pgno == lockPgno {
    continue // Skip this page - it's reserved by SQLite
}

type DB struct {
    path     string      // Database file path
    db       *sql.DB     // SQLite connection
    rtx      *sql.Tx     // Long-running read transaction
    pageSize int         // Database page size (critical for lock page)
    notify   chan struct{} // Notifies on WAL changes
}

type ReplicaClient interface {
    Type() string  // Client type identifier

    // File operations
    LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)
    OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)
    WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error)
    DeleteLTXFiles(ctx context.Context, files []*ltx.FileInfo) error
    DeleteAll(ctx context.Context) error
}

type Compactor struct {
    client           ReplicaClient
    VerifyCompaction bool  // Post-compaction TXID consistency check
    RetentionEnabled bool  // Default: true. Controls active file deletion

    // Local file optimization (set by DB layer)
    LocalFileOpener  func(level int, minTXID, maxTXID ltx.TXID) (io.ReadCloser, error)
    LocalFileDeleter func(level int, minTXID, maxTXID ltx.TXID) error

    // Level max-file-info caching
    CacheGetter func(level int) (*ltx.FileInfo, bool)
    CacheSetter func(level int, info *ltx.FileInfo)
}

var defaultLevels = CompactionLevels{
    {Level: 0, Interval: 0},        // Raw LTX files (no compaction)
    {Level: 1, Interval: 30*Second},
    {Level: 2, Interval: 5*Minute},
    {Level: 3, Interval: 1*Hour},
    // Snapshots created daily (24h retention)
}

# Always run with race detector
go test -race -v ./...

# Specific race-prone areas
go test -race -v -run TestReplica_Sync ./...
go test -race -v -run TestDB_Sync ./...
go test -race -v -run TestStore_CompactDB ./...

# Test with various page sizes
./bin/litestream-test populate -db test.db -page-size 4096 -target-size 2GB
./bin/litestream-test populate -db test.db -page-size 8192 -target-size 2GB

# Validate lock page handling
./bin/litestream-test validate -source-db test.db -replica-url file:///tmp/replica

# Test specific backend
go test -v ./replica_client_test.go -integration s3
go test -v ./replica_client_test.go -integration gcs
go test -v ./replica_client_test.go -integration abs
go test -v ./replica_client_test.go -integration oss
go test -v ./replica_client_test.go -integration sftp

replicas:
  - url: s3://bucket-name/path
    region: us-east-1

replicas:
  - url: s3://bucket-name/path?endpoint=https://ACCOUNT_ID.r2.cloudflarestorage.com
    access-key-id: your-access-key-id
    secret-access-key: your-secret-access-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://s3.REGION.backblazeb2.com&sign-payload=true&force-path-style=true
    access-key-id: your-key-id
    secret-access-key: your-application-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://REGION.digitaloceanspaces.com&force-path-style=false
    access-key-id: your-spaces-key
    secret-access-key: your-spaces-secret

replicas:
  - url: s3://bucket-name/path?endpoint=https://your-minio-server:9000&force-path-style=true
    access-key-id: your-access-key
    secret-access-key: your-secret-key

dbs:
  - path: /path/to/database.db
    replica:
      type: s3
      bucket: bucket-name
      path: database-name
      endpoint: https://your-garage-server
      region: garage
      force-path-style: true
      sign-accept-encoding: false
      access-key-id: your-access-key
      secret-access-key: your-secret-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://s3.REGION.scw.cloud&force-path-style=true
    access-key-id: your-access-key
    secret-access-key: your-secret-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://REGION.your-objectstorage.com&force-path-style=true
    access-key-id: your-access-key
    secret-access-key: your-secret-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://s3.filebase.com&force-path-style=true
    access-key-id: your-access-key
    secret-access-key: your-secret-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://fly.storage.tigris.dev&force-path-style=true
    access-key-id: your-access-key
    secret-access-key: your-secret-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://PROJECT_REF.supabase.co/storage/v1/s3
    access-key-id: your-s3-access-key
    secret-access-key: your-s3-secret-key

replicas:
  - url: s3://bucket-name/path?endpoint=https://s3.REGION.wasabisys.com
    access-key-id: your-access-key
    secret-access-key: your-secret-key

replicas:
  - url: gcs://bucket-name/path

replicas:
  - type: s3
    bucket: bucket-name
    path: database-name
    endpoint: https://storage.googleapis.com
    region: us-east-1
    force-path-style: true
    access-key-id: your-hmac-access-key
    secret-access-key: your-hmac-secret

replicas:
  - url: abs://container-name/path
    account-name: your-account-name
    account-key: your-account-key

replicas:
  - url: abs://container-name/path
    account-name: your-account-name
    sas-token: "sv=2023-01-03&ss=b&srt=co&sp=rwdlacx..."

replicas:
  - url: oss://bucket-name/path?endpoint=oss-REGION.aliyuncs.com
    access-key-id: your-access-key-id
    access-key-secret: your-access-key-secret

replicas:
  - url: sftp://hostname/path
    user: username
    password: password  # or use key-path

LITESTREAM_DEBUG=1 litestream replicate ...

logging:
  level: debug

# List any existing backups
litestream snapshots s3://bucket/path?endpoint=...

# Perform a test restore (requires existing backup)
litestream restore -o /tmp/test.db s3://bucket/path?endpoint=...

validation:
  interval: "5m"  # How often to validate; 0 disables

type ReplicaClient interface {
    // Returns the type identifier (e.g., "s3", "gcs", "file")
    Type() string

    // Returns iterator of LTX files at given level
    // seek: Start from this TXID (0 = beginning)
    // useMetadata: When true, fetch accurate timestamps from backend metadata (required for PIT restore)
    LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)

    // Opens an LTX file for reading
    // Returns os.ErrNotExist if file doesn't exist
    OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)

    // Writes an LTX file to storage
    // SHOULD set CreatedAt based on backend metadata or upload time
    WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error)

    // Deletes one or more LTX files
    DeleteLTXFiles(ctx context.Context, files []*ltx.FileInfo) error

    // Deletes all files for this database
    DeleteAll(ctx context.Context) error
}

/* Detailed source-code truncated for AI context efficiency. */

func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
    // List files from storage
    files, err := c.listFiles(ctx, level, useMetadata)
    if err != nil {
        return nil, err
    }

    // Sort by TXID for consistent ordering
    sort.Slice(files, func(i, j int) bool {
        if files[i].MinTXID != files[j].MinTXID {
            return files[i].MinTXID < files[j].MinTXID
        }
        return files[i].MaxTXID < files[j].MaxTXID
    })

    // Filter by seek position
    var filtered []*ltx.FileInfo
    for _, f := range files {
        if f.MinTXID >= seek {
            filtered = append(filtered, f)
        }
    }

    return ltx.NewFileInfoSliceIterator(filtered), nil
}

func (c *ReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
    path := c.ltxPath(level, minTXID, maxTXID)

    // For eventually consistent backends, implement retry
    var lastErr error
    backoff := 100 * time.Millisecond

    for i := 0; i < 5; i++ {
        reader, err := c.openFile(ctx, path, offset, size)
        if err == nil {
            return reader, nil
        }

        // Don't retry on definitive errors
        if errors.Is(err, os.ErrNotExist) {
            return nil, err
        }

        lastErr = err
        select {
        case <-ctx.Done():
            return nil, ctx.Err()
        case <-time.After(backoff):
            backoff *= 2
        }
    }

    return nil, fmt.Errorf("open file after retries: %w", lastErr)
}

// File not found
return nil, os.ErrNotExist

// Permission denied
return nil, os.ErrPermission

// Context cancelled
return nil, ctx.Err()

// Custom errors should wrap standard ones
return nil, fmt.Errorf("s3 download failed: %w", err)

// Retryable errors
func isRetryable(err error) bool {
    // Network errors
    var netErr net.Error
    if errors.As(err, &netErr) && netErr.Temporary() {
        return true
    }

    // Specific HTTP status codes
    if httpErr, ok := err.(HTTPError); ok {
        switch httpErr.StatusCode {
        case 429, 500, 502, 503, 504:
            return true
        }
    }

    // Timeout errors
    if errors.Is(err, context.DeadlineExceeded) {
        return true
    }

    return false
}

func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
    logger := slog.Default().With(
        "replica", c.Type(),
        "level", level,
        "minTXID", minTXID,
        "maxTXID", maxTXID,
    )

    logger.Debug("starting ltx upload")

    info, err := c.upload(ctx, level, minTXID, maxTXID, r)
    if err != nil {
        logger.Error("ltx upload failed", "error", err)
        return nil, err
    }

    logger.Info("ltx upload complete", "size", info.Size)
    return info, nil
}

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

// mock/replica_client.go
type ReplicaClient struct {
    mu     sync.Mutex
    files  map[string]*ltx.FileInfo
    data   map[string][]byte
    errors map[string]error  // Inject errors for testing
}

func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Check for injected error
    key := fmt.Sprintf("write-%d-%d-%d", level, minTXID, maxTXID)
    if err, ok := c.errors[key]; ok {
        return nil, err
    }

    // Store data
    data, err := io.ReadAll(r)
    if err != nil {
        return nil, err
    }

    path := ltxPath(level, minTXID, maxTXID)
    c.data[path] = data

    info := &ltx.FileInfo{
        Level:     level,
        MinTXID:   minTXID,
        MaxTXID:   maxTXID,
        Size:      int64(len(data)),
        CreatedAt: time.Now(),
    }
    c.files[path] = info

    return info, nil
}

type ReplicaClientV3 interface {
    GenerationsV3(ctx context.Context) ([]string, error)
    SnapshotsV3(ctx context.Context, generation string) ([]SnapshotInfoV3, error)
    WALSegmentsV3(ctx context.Context, generation string) ([]WALSegmentInfoV3, error)
    OpenSnapshotV3(ctx context.Context, generation string, index int) (io.ReadCloser, error)
    OpenWALSegmentV3(ctx context.Context, generation string, index int, offset int64) (io.ReadCloser, error)
}

type PosV3 struct {
    Generation string // 16-char hex string
    Index      int    // WAL index
    Offset     int64  // Offset within WAL segment
}

type SnapshotInfoV3 struct {
    Generation string
    Index      int
    Size       int64
    CreatedAt  time.Time
}

type WALSegmentInfoV3 struct {
    Generation string
    Index      int
    Offset     int64
    Size       int64
    CreatedAt  time.Time
}

generations/
  {generation-id}/         # 16-char hex (validated by IsGenerationIDV3)
    snapshots/
      {index:08x}.snapshot.lz4
    wal/
      {index:08x}_{offset:08x}.wal.lz4

// WRONG - Always reads entire file
func (c *Client) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
    return c.storage.Download(path)  // Ignores offset/size!
}

// CORRECT - Respects offset and size
func (c *Client) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
    if offset == 0 && size == 0 {
        // Full file
        return c.storage.Download(path)
    }

    // Partial read using Range header or equivalent
    end := offset + size - 1
    if size == 0 {
        end = 0  // Read to end
    }
    return c.storage.DownloadRange(path, offset, end)
}

// WRONG - Uses current time
func (c *Client) WriteLTXFile(...) (*ltx.FileInfo, error) {
    // Upload file...

    return &ltx.FileInfo{
        CreatedAt: time.Now(),  // Wrong! Loses temporal info
    }, nil
}

// CORRECT - Preserves original timestamp
func (c *Client) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
    // Upload file...
    uploadedSize, modTime, err := c.storage.Upload(path, r)
    if err != nil {
        return nil, err
    }

    return &ltx.FileInfo{
        Level:     level,
        MinTXID:   minTXID,
        MaxTXID:   maxTXID,
        Size:      uploadedSize,
        CreatedAt: modTime,
    }, nil
}

// WRONG - Generic error
func (c *Client) OpenLTXFile(...) (io.ReadCloser, error) {
    resp, err := c.get(path)
    if err != nil {
        return nil, fmt.Errorf("not found")  // Wrong type!
    }
}

// CORRECT - Proper error type
func (c *Client) OpenLTXFile(...) (io.ReadCloser, error) {
    resp, err := c.get(path)
    if err != nil {
        if resp.StatusCode == 404 {
            return nil, os.ErrNotExist  // Correct type
        }
        return nil, fmt.Errorf("download failed: %w", err)
    }
}

// WRONG - Ignores context
func (c *Client) WriteLTXFile(ctx context.Context, ...) (*ltx.FileInfo, error) {
    // Long operation without checking context
    for i := 0; i < 1000000; i++ {
        doWork()  // Could run forever!
    }
}

// CORRECT - Respects context
func (c *Client) WriteLTXFile(ctx context.Context, ...) (*ltx.FileInfo, error) {
    // Check context periodically
    for i := 0; i < 1000000; i++ {
        select {
        case <-ctx.Done():
            return nil, ctx.Err()
        default:
            // Continue work
        }

        if err := doWork(ctx); err != nil {
            return nil, err
        }
    }
}

// WRONG - Loads all files at once
func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
    allFiles, err := c.loadAllFiles(level)  // Could be millions!
    if err != nil {
        return nil, err
    }

    return NewIterator(allFiles), nil
}

// CORRECT - Lazy loading with pagination
func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
    return &lazyIterator{
        client:      c,
        level:       level,
        seek:        seek,
        useMetadata: useMetadata,
        pageSize:    1000,
    }, nil
}

type lazyIterator struct {
    client   *Client
    level    int
    seek     ltx.TXID
    pageSize int
    current  []*ltx.FileInfo
    index    int
    done     bool
}

func (i *lazyIterator) Next() bool {
    if i.index >= len(i.current) && !i.done {
        // Load next page
        i.loadNextPage()
    }
    return i.index < len(i.current)
}

// Path construction
func (c *ReplicaClient) ltxDir(level int) string {
    if level == SnapshotLevel {
        return path.Join(c.Path, "snapshots")
    }
    return path.Join(c.Path, "ltx", fmt.Sprintf("%04d", level))
}

// Metadata handling
func (c *ReplicaClient) WriteLTXFile(...) (*ltx.FileInfo, error) {
    // Add metadata to object
    metadata := map[string]string{
        "min-txid": fmt.Sprintf("%d", minTXID),
        "max-txid": fmt.Sprintf("%d", maxTXID),
        "level":    fmt.Sprintf("%d", level),
    }

    // Upload with metadata
    _, err := c.s3.PutObjectWithContext(ctx, &s3.PutObjectInput{
        Bucket:   &c.Bucket,
        Key:      &key,
        Body:     r,
        Metadata: metadata,
    })
}

// Error mapping
func mapS3Error(err error) error {
    if aerr, ok := err.(awserr.Error); ok {
        switch aerr.Code() {
        case s3.ErrCodeNoSuchKey:
            return os.ErrNotExist
        case s3.ErrCodeAccessDenied:
            return os.ErrPermission
        }
    }
    return err
}

type ReplicaClient struct {
    pool *ConnectionPool
}

func NewReplicaClient(config Config) *ReplicaClient {
    pool := &ConnectionPool{
        MaxConnections: config.MaxConnections,
        IdleTimeout:    config.IdleTimeout,
    }

    return &ReplicaClient{
        pool: pool,
    }
}

func (c *ReplicaClient) DeleteLTXFiles(ctx context.Context, files []*ltx.FileInfo) error {
    // Batch deletes for efficiency
    const batchSize = 100

    for i := 0; i < len(files); i += batchSize {
        end := i + batchSize
        if end > len(files) {
            end = len(files)
        }

        batch := files[i:end]
        if err := c.deleteBatch(ctx, batch); err != nil {
            return fmt.Errorf("delete batch %d: %w", i/batchSize, err)
        }
    }

    return nil
}

type ReplicaClient struct {
    cache *FileInfoCache
}

func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
    // Check cache first (only cache when useMetadata=false for fast queries)
    cacheKey := fmt.Sprintf("%d-%d", level, seek)
    if !useMetadata {
        if cached, ok := c.cache.Get(cacheKey); ok {
            return ltx.NewFileInfoSliceIterator(cached), nil
        }
    }

    // Load from storage
    files, err := c.loadFiles(ctx, level, seek, useMetadata)
    if err != nil {
        return nil, err
    }

    // Cache for future requests
    c.cache.Set(cacheKey, files, 5*time.Minute)

    return ltx.NewFileInfoSliceIterator(files), nil
}

database.db         # Main database file (pages)
database.db-wal     # Write-ahead log
database.db-shm     # Shared memory file

graph LR
    Write[Write Transaction] -->|Append| WAL[WAL File]
    WAL -->|Checkpoint| DB[Main Database]
    Read[Read Transaction] -->|Merge View| View[Consistent View]
    DB --> View
    WAL --> View

+------------------+
| WAL Header       | 32 bytes
+------------------+
| Frame 1 Header   | 24 bytes
| Frame 1 Data     | Page size bytes
+------------------+
| Frame 2 Header   | 24 bytes
| Frame 2 Data     | Page size bytes
+------------------+
| ...              |
+------------------+

type WALHeader struct {
    Magic         [4]byte  // 0x377f0682 or 0x377f0683
    FileFormat    uint32   // File format version (3007000)
    PageSize      uint32   // Database page size
    Checkpoint    uint32   // Checkpoint sequence number
    Salt1         uint32   // Random salt for checksum
    Salt2         uint32   // Random salt for checksum
    Checksum1     uint32   // Header checksum
    Checksum2     uint32   // Header checksum
}

type WALFrameHeader struct {
    PageNumber    uint32   // Page number in database
    DbSize        uint32   // Size of database in pages
    Salt1         uint32   // Must match header salt
    Salt2         uint32   // Must match header salt
    Checksum1     uint32   // Cumulative checksum
    Checksum2     uint32   // Cumulative checksum
}

// db.go - Reading WAL for replication
func (db *DB) readWAL() ([]Page, error) {
    walPath := db.path + "-wal"
    f, err := os.Open(walPath)
    if err != nil {
        return nil, err
    }
    defer f.Close()

    // Read WAL header
    var header WALHeader
    binary.Read(f, binary.BigEndian, &header)

    // Validate magic number
    magic := binary.BigEndian.Uint32(header.Magic[:])
    if magic != 0x377f0682 && magic != 0x377f0683 {
        return nil, errors.New("invalid WAL magic")
    }

    // Read frames
    var pages []Page
    for {
        var frameHeader WALFrameHeader
        err := binary.Read(f, binary.BigEndian, &frameHeader)
        if err == io.EOF {
            break
        }

        // Read page data
        pageData := make([]byte, header.PageSize)
        f.Read(pageData)

        pages = append(pages, Page{
            Number: frameHeader.PageNumber,
            Data:   pageData,
        })
    }

    return pages, nil
}

Page Size: Typically 4096 bytes (4KB)
Page Number: 1-based indexing
Page Types:
  - B-tree interior pages
  - B-tree leaf pages
  - Overflow pages
  - Freelist pages
  - Lock byte page (at 1GB)

+------------------+
| Page Header      | Variable (8-12 bytes)
+------------------+
| Cell Pointers    | 2 bytes each
+------------------+
| Unallocated      |
| Space            |
+------------------+
| Cell Content     | Variable size
| Area             | (grows upward)
+------------------+

type PageHeader struct {
    PageType      byte     // 0x02, 0x05, 0x0a, 0x0d
    FreeBlockStart uint16  // Start of free block list
    CellCount     uint16   // Number of cells
    CellStart     uint16   // Offset to first cell
    FragmentBytes byte     // Fragmented free bytes
    // Additional fields for interior pages
    RightChild    uint32   // Only for interior pages
}

const PENDING_BYTE = 0x40000000  // 1GB mark

// Page number varies by page size
func LockPageNumber(pageSize int) uint32 {
    return uint32(PENDING_BYTE/pageSize) + 1
}

// Examples:
// 4KB pages:  262145 (0x40001)
// 8KB pages:  131073 (0x20001)
// 16KB pages:  65537 (0x10001)
// 32KB pages:  32769 (0x08001)
// 64KB pages:  16385 (0x04001)

// From superfly/ltx package
func LockPgno(pageSize int) uint32 {
    return uint32(PENDING_BYTE/pageSize) + 1
}

// db.go - Skipping lock page during iteration
for pgno := uint32(1); pgno <= maxPgno; pgno++ {
    if pgno == ltx.LockPgno(db.pageSize) {
        continue // Skip lock page
    }

    // Process normal page
    processPage(pgno)
}

-- Create database that spans lock page
CREATE TABLE test (id INTEGER PRIMARY KEY, data BLOB);

-- Insert data until database > 1GB
WITH RECURSIVE generate_series(value) AS (
    SELECT 1
    UNION ALL
    SELECT value+1 FROM generate_series
    LIMIT 300000
)
INSERT INTO test SELECT value, randomblob(4000) FROM generate_series;

-- Check database size
PRAGMA page_count;  -- Should be > 262145 for 4KB pages
PRAGMA page_size;   -- Typically 4096

-- Calculate if lock page is in range
-- For 4KB pages: if page_count > 262145, lock page is included

graph TD
    UNLOCKED -->|BEGIN| SHARED
    SHARED -->|Write| RESERVED
    RESERVED -->|Prepare| PENDING
    PENDING -->|Commit| EXCLUSIVE
    EXCLUSIVE -->|Done| UNLOCKED

// db.go - Maintaining read transaction for consistency
func (db *DB) initReadTx() error {
    // Start read-only transaction
    tx, err := db.db.BeginTx(context.Background(), &sql.TxOptions{
        ReadOnly: true,
    })
    if err != nil {
        return err
    }

    // Execute query to acquire SHARED lock
    var dummy string
    err = tx.QueryRow("SELECT ''").Scan(&dummy)
    if err != nil {
        tx.Rollback()
        return err
    }

    // Keep transaction open
    db.rtx = tx
    return nil
}

PRAGMA wal_checkpoint(PASSIVE);

PRAGMA wal_checkpoint(FULL);

PRAGMA wal_checkpoint(RESTART);

PRAGMA wal_checkpoint(TRUNCATE);

// db.go - Checkpoint decision logic
func (db *DB) autoCheckpoint() error {
    walSize := db.WALSize()
    pageCount := walSize / db.pageSize

    if pageCount > db.TruncatePageN {
        // Force truncation for very large WAL (emergency brake)
        return db.Checkpoint("TRUNCATE")
    } else if pageCount > db.MinCheckpointPageN {
        // Try passive checkpoint (non-blocking)
        return db.Checkpoint("PASSIVE")
    } else if db.CheckpointInterval elapsed {
        // Time-based passive checkpoint
        return db.Checkpoint("PASSIVE")
    }
    // Note: RESTART mode permanently removed due to issue #724 (write-blocking)
    }

    return nil
}

-- Enable WAL mode (required)
PRAGMA journal_mode = WAL;

-- Get database info
PRAGMA page_size;        -- Page size in bytes
PRAGMA page_count;       -- Total pages in database
PRAGMA freelist_count;   -- Free pages

-- WAL information
PRAGMA wal_checkpoint;           -- Perform checkpoint
PRAGMA wal_autocheckpoint;       -- Auto-checkpoint threshold
PRAGMA wal_checkpoint(PASSIVE);  -- Non-blocking checkpoint

-- Database state
PRAGMA integrity_check;  -- Verify database integrity
PRAGMA quick_check;     -- Fast integrity check

-- Lock information
PRAGMA lock_status;     -- Current locks (debug builds)

-- Performance tuning
PRAGMA synchronous = NORMAL;  -- Sync mode
PRAGMA busy_timeout = 5000;   -- Wait 5s for locks
PRAGMA cache_size = -64000;   -- 64MB cache

func getDatabaseInfo(db *sql.DB) (*DBInfo, error) {
    info := &DBInfo{}

    // Page size
    err := db.QueryRow("PRAGMA page_size").Scan(&info.PageSize)

    // Page count
    err = db.QueryRow("PRAGMA page_count").Scan(&info.PageCount)

    // Journal mode
    err = db.QueryRow("PRAGMA journal_mode").Scan(&info.JournalMode)

    // Calculate size
    info.Size = info.PageSize * info.PageCount

    return info, nil
}

// Using database/sql for queries
db, err := sql.Open("sqlite3", "database.db")

// Using modernc.org/sqlite for low-level access
conn, err := sqlite.Open("database.db")

// Direct page access (requires special builds)
page := readPage(conn, pageNumber)

// Read-only connection
db, err := sql.Open("sqlite3", "file:database.db?mode=ro")

// WAL mode connection
db, err := sql.Open("sqlite3", "database.db?_journal=WAL")

// With busy timeout
db, err := sql.Open("sqlite3", "database.db?_busy_timeout=5000")

// Multiple options
db, err := sql.Open("sqlite3", "database.db?_journal=WAL&_busy_timeout=5000&_synchronous=NORMAL")

// Litestream's approach to reading WAL
func (db *DB) monitorWAL() {
    walPath := db.path + "-wal"

    for {
        // Check WAL file size
        stat, err := os.Stat(walPath)
        if err != nil {
            continue // WAL might not exist yet
        }

        // Compare with last known size
        if stat.Size() > db.lastWALSize {
            // New data in WAL
            db.processWALChanges()
            db.lastWALSize = stat.Size()
        }

        time.Sleep(db.MonitorInterval)
    }
}

// Can interfere with Litestream's control
// Solution: Set high threshold
db.Exec("PRAGMA wal_autocheckpoint = 10000")

// Set reasonable timeout
db.Exec("PRAGMA busy_timeout = 5000") // 5 seconds

// NORMAL is safe with WAL
db.Exec("PRAGMA synchronous = NORMAL")

// Set cache size (negative = KB, positive = pages)
db.Exec("PRAGMA cache_size = -64000") // 64MB

func walToLTX(walFrames []WALFrame) *LTXFile {
    ltx := &LTXFile{
        Header: LTXHeader{
            PageSize: walFrames[0].PageSize,
            MinTXID:  walFrames[0].TransactionID,
        },
    }

    for _, frame := range walFrames {
        // Skip lock page
        if frame.PageNumber == LockPageNumber(ltx.Header.PageSize) {
            continue
        }

        ltx.Pages = append(ltx.Pages, Page{
            Number: frame.PageNumber,
            Data:   frame.Data,
        })

        ltx.Header.MaxTXID = frame.TransactionID
    }

    return ltx
}

# Use litestream-test tool for large databases
./bin/litestream-test populate \
    -db test.db \
    -target-size 1.5GB \
    -page-size 4096

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

# Always run tests with race detector
go test -race -v ./...

# Run specific race-prone tests
go test -race -v -run TestReplica_Sync ./...
go test -race -v -run TestDB_Sync ./...
go test -race -v -run TestStore_CompactDB ./...

func TestReplica_ConcurrentPositionUpdate(t *testing.T) {
    replica := litestream.NewReplica(nil)
    var wg sync.WaitGroup
    errors := make(chan error, 100)

    // Concurrent writers
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()

            pos := ltx.NewPos(ltx.TXID(n), ltx.Checksum(uint64(n)))

            // This should use proper locking
            replica.SetPos(pos)

            // Verify position
            readPos := replica.Pos()
            if readPos.TXID < ltx.TXID(n) {
                errors <- fmt.Errorf("position went backwards")
            }
        }(i)
    }

    // Concurrent readers
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()

            for j := 0; j < 100; j++ {
                _ = replica.Pos()
                time.Sleep(time.Microsecond)
            }
        }()
    }

    wg.Wait()
    close(errors)

    for err := range errors {
        t.Error(err)
    }
}

func TestDB_ConcurrentWALAccess(t *testing.T) {
    db := setupTestDB(t)
    defer db.Close(context.Background())

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    var wg sync.WaitGroup

    // Writer goroutine
    wg.Add(1)
    go func() {
        defer wg.Done()

        conn, err := sql.Open("sqlite", db.Path())
        if err != nil {
            return
        }
        defer conn.Close()

        for i := 0; i < 100; i++ {
            _, _ = conn.Exec("INSERT INTO test VALUES (?)", i)
            time.Sleep(10 * time.Millisecond)
        }
    }()

    // Monitor goroutine
    wg.Add(1)
    go func() {
        defer wg.Done()

        notifyCh := db.Notify()

        for {
            select {
            case <-ctx.Done():
                return
            case <-notifyCh:
                // Process WAL changes
                _ = db.Sync(context.Background())
            }
        }
    }()

    // Checkpoint goroutine
    wg.Add(1)
    go func() {
        defer wg.Done()

        ticker := time.NewTicker(100 * time.Millisecond)
        defer ticker.Stop()

        for {
            select {
            case <-ctx.Done():
                return
            case <-ticker.C:
                _ = db.Checkpoint(context.Background(), litestream.CheckpointModePassive)
            }
        }
    }()

    wg.Wait()
}

func TestStore_Integration(t *testing.T) {
    // Setup
    tmpDir := t.TempDir()
    db := setupTestDB(t, tmpDir)

    // Use defer with error channel for cleanup
    insertErr := make(chan error, 1)

    // Cleanup function
    cleanup := func() {
        select {
        case err := <-insertErr:
            if err != nil {
                t.Errorf("insert error during test: %v", err)
            }
        default:
        }
    }
    defer cleanup()

    // Test with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // Run test...
}

// +build integration

package litestream_test

import (
    "context"
    "os"
    "testing"
)

func TestIntegration_S3(t *testing.T) {
    if testing.Short() {
        t.Skip("skipping integration test")
    }

    // Check for credentials
    if os.Getenv("AWS_ACCESS_KEY_ID") == "" {
        t.Skip("AWS_ACCESS_KEY_ID not set")
    }

    // Run test against real S3
    runIntegrationTest(t, setupS3Client())
}

func runIntegrationTest(t *testing.T, client ReplicaClient) {
    ctx := context.Background()

    t.Run("BasicReplication", func(t *testing.T) {
        // Test basic write/read cycle
    })

    t.Run("Compaction", func(t *testing.T) {
        // Test compaction with remote storage
    })

    t.Run("EventualConsistency", func(t *testing.T) {
        // Test handling of eventual consistency
    })

    t.Run("LargeFiles", func(t *testing.T) {
        // Test with files > 100MB
    })

    t.Run("Cleanup", func(t *testing.T) {
        err := client.DeleteAll(ctx)
        require.NoError(t, err)
    })
}

func setupS3Client() *s3.ReplicaClient {
    return &s3.ReplicaClient{
        AccessKeyID:     os.Getenv("AWS_ACCESS_KEY_ID"),
        SecretAccessKey: os.Getenv("AWS_SECRET_ACCESS_KEY"),
        Region:          getEnvOrDefault("AWS_REGION", "us-east-1"),
        Bucket:          getEnvOrDefault("TEST_S3_BUCKET", "litestream-test"),
        Path:            fmt.Sprintf("test-%d", time.Now().Unix()),
    }
}

func getEnvOrDefault(key, defaultValue string) string {
    if value := os.Getenv(key); value != "" {
        return value
    }
    return defaultValue
}

func BenchmarkDB_Sync(b *testing.B) {
    db := setupBenchDB(b)
    defer db.Close(context.Background())

    // Prepare test data
    conn, _ := sql.Open("sqlite", db.Path())
    defer conn.Close()

    for i := 0; i < 1000; i++ {
        conn.Exec("INSERT INTO test VALUES (?)", i)
    }

    b.ResetTimer()

    for i := 0; i < b.N; i++ {
        err := db.Sync(context.Background())
        if err != nil {
            b.Fatal(err)
        }
    }

    b.ReportMetric(float64(b.N)/b.Elapsed().Seconds(), "syncs/sec")
}

func BenchmarkCompaction(b *testing.B) {
    benchmarks := []struct {
        name      string
        fileCount int
        fileSize  int
    }{
        {"Small-Many", 1000, 1024},        // Many small files
        {"Medium", 100, 10 * 1024},        // Medium files
        {"Large-Few", 10, 100 * 1024},     // Few large files
    }

    for _, bm := range benchmarks {
        b.Run(bm.name, func(b *testing.B) {
            for i := 0; i < b.N; i++ {
                b.StopTimer()
                files := generateTestFiles(bm.fileCount, bm.fileSize)
                b.StartTimer()

                _, err := compact(files)
                if err != nil {
                    b.Fatal(err)
                }
            }

            totalSize := int64(bm.fileCount * bm.fileSize)
            b.ReportMetric(float64(totalSize)/float64(b.Elapsed().Nanoseconds()), "bytes/ns")
        })
    }
}

/* Detailed source-code truncated for AI context efficiency. */

type MockReplicaClient struct {
    mu    sync.Mutex
    files map[string]*ltx.FileInfo
    data  map[string][]byte

    // Control behavior
    FailureRate   float64
    Latency       time.Duration
    EventualDelay time.Duration
}

func (m *MockReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
    // Simulate latency
    if m.Latency > 0 {
        time.Sleep(m.Latency)
    }

    // Simulate failures
    if m.FailureRate > 0 && rand.Float64() < m.FailureRate {
        return nil, errors.New("simulated failure")
    }

    // Simulate eventual consistency
    if m.EventualDelay > 0 {
        time.AfterFunc(m.EventualDelay, func() {
            m.mu.Lock()
            defer m.mu.Unlock()
            // Make file available after delay
        })
    }

    // Store file
    data, err := io.ReadAll(r)
    if err != nil {
        return nil, err
    }

    m.mu.Lock()
    defer m.mu.Unlock()

    key := fmt.Sprintf("%d-%016x-%016x", level, minTXID, maxTXID)
    info := &ltx.FileInfo{
        Level:     level,
        MinTXID:   minTXID,
        MaxTXID:   maxTXID,
        Size:      int64(len(data)),
        CreatedAt: time.Now(),
    }

    m.files[key] = info
    m.data[key] = data

    return info, nil
}

type MockDB struct {
    mu       sync.Mutex
    path     string
    replicas []*Replica
    closed   bool

    // Control behavior
    CheckpointFailures int
    SyncDelay         time.Duration
}

func (m *MockDB) Sync(ctx context.Context) error {
    if m.SyncDelay > 0 {
        select {
        case <-time.After(m.SyncDelay):
        case <-ctx.Done():
            return ctx.Err()
        }
    }

    m.mu.Lock()
    defer m.mu.Unlock()

    if m.closed {
        return errors.New("database closed")
    }

    for _, r := range m.replicas {
        if err := r.Sync(ctx); err != nil {
            return err
        }
    }

    return nil
}

// testutil/db.go
package testutil

import (
    "database/sql"
    "testing"
    "path/filepath"
)

func NewTestDB(t testing.TB) *litestream.DB {
    t.Helper()

    path := filepath.Join(t.TempDir(), "test.db")

    // Create SQLite database
    conn, err := sql.Open("sqlite", path+"?_journal=WAL")
    require.NoError(t, err)

    _, err = conn.Exec(`
        CREATE TABLE test (
            id INTEGER PRIMARY KEY,
            data BLOB
        )
    `)
    require.NoError(t, err)
    conn.Close()

    // Open with Litestream
    db := litestream.NewDB(path)
    db.MonitorInterval = 10 * time.Millisecond  // Speed up for tests
    db.MinCheckpointPageN = 100  // Lower threshold for tests

    err = db.Open()
    require.NoError(t, err)

    t.Cleanup(func() {
        db.Close(context.Background())
    })

    return db
}

func WriteTestData(t testing.TB, db *litestream.DB, count int) {
    t.Helper()

    conn, err := sql.Open("sqlite", db.Path())
    require.NoError(t, err)
    defer conn.Close()

    tx, err := conn.Begin()
    require.NoError(t, err)

    for i := 0; i < count; i++ {
        data := make([]byte, 100)
        rand.Read(data)
        _, err = tx.Exec("INSERT INTO test (data) VALUES (?)", data)
        require.NoError(t, err)
    }

    err = tx.Commit()
    require.NoError(t, err)
}

// testdata/fixtures.go
package testdata

import _ "embed"

//go:embed small.db
var SmallDB []byte

//go:embed large.db
var LargeDB []byte

//go:embed corrupted.db
var CorruptedDB []byte

func ExtractFixture(name string, path string) error {
    var data []byte

    switch name {
    case "small":
        data = SmallDB
    case "large":
        data = LargeDB
    case "corrupted":
        data = CorruptedDB
    default:
        return fmt.Errorf("unknown fixture: %s", name)
    }

    return os.WriteFile(path, data, 0600)
}

// Problem: Multiple connections without proper WAL mode
func TestBroken(t *testing.T) {
    db1, _ := sql.Open("sqlite", "test.db")    // Wrong! WAL disabled
    db2, _ := sql.Open("sqlite", "test.db")    // Will fail
}

// Solution: Use WAL mode
func TestFixed(t *testing.T) {
    db1, _ := sql.Open("sqlite", "test.db?_journal=WAL")
    db2, _ := sql.Open("sqlite", "test.db?_journal=WAL")
}

// Problem: Race between write and sync
func TestBroken(t *testing.T) {
    WriteData(db)
    result := ReadReplica()  // May not see data yet!
}

// Solution: Explicit sync
func TestFixed(t *testing.T) {
    WriteData(db)
    err := db.Sync(context.Background())
    require.NoError(t, err)
    result := ReadReplica()  // Now guaranteed to see data
}

// Problem: Goroutine outlives test
func TestBroken(t *testing.T) {
    go func() {
        time.Sleep(10 * time.Second)
        doWork()  // Test already finished!
    }()
}

// Solution: Use context and wait
func TestFixed(t *testing.T) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        select {
        case <-ctx.Done():
            return
        case <-time.After(10 * time.Second):
            doWork()
        }
    }()

    // Test work...

    cancel()  // Signal shutdown
    wg.Wait() // Wait for goroutine
}

// Problem: Not closing files
func TestBroken(t *testing.T) {
    f, _ := os.Open("test.db")
    // Missing f.Close()!
}

// Solution: Always use defer
func TestFixed(t *testing.T) {
    f, err := os.Open("test.db")
    require.NoError(t, err)
    defer f.Close()
}

# Generate coverage report
go test -coverprofile=coverage.out ./...

# View coverage in browser
go tool cover -html=coverage.out

# Check coverage percentage
go tool cover -func=coverage.out | grep total

# Coverage by package
go test -cover ./...

// Use test tables for comprehensive coverage
func TestDB_Checkpoint(t *testing.T) {
    tests := []struct {
        name     string
        mode     string
        walSize  int
        wantErr  bool
    }{
        {"Passive", "PASSIVE", 100, false},
        {"Full", "FULL", 1000, false},
        {"Restart", "RESTART", 5000, false},
        {"Truncate", "TRUNCATE", 10000, false},
        {"Invalid", "INVALID", 100, true},
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            db := setupTestDB(t)
            generateWAL(t, db, tt.walSize)

            err := db.Checkpoint(tt.mode)
            if tt.wantErr {
                assert.Error(t, err)
            } else {
                assert.NoError(t, err)
            }
        })
    }
}

make vfs

# macOS ARM64 (Apple Silicon)
make vfs-darwin-arm64
# Output: dist/litestream-vfs-darwin-arm64.dylib

# macOS AMD64 (Intel)
make vfs-darwin-amd64
# Output: dist/litestream-vfs-darwin-amd64.dylib

# Linux AMD64
make vfs-linux-amd64
# Output: dist/litestream-vfs-linux-amd64.so

# Linux ARM64
make vfs-linux-arm64
# Output: dist/litestream-vfs-linux-arm64.so

make vfs-test

# Amazon S3
s3://bucket-name/path/to/db

# Google Cloud Storage
gs://bucket-name/path/to/db

# Azure Blob Storage
abs://container-name/path/to/db

# Alibaba OSS
oss://bucket-name/path/to/db

# Local filesystem
file:///path/to/replica

# SFTP
sftp://user@host:port/path/to/db

# NATS JetStream
nats://host:port/bucket/path

# WebDAV
webdav://host:port/path/to/db

-- Load the extension (adjust path as needed)
.load ./dist/litestream-vfs.so

export LITESTREAM_REPLICA_URL="s3://my-bucket/mydb"

.load ./dist/litestream-vfs.so
.open file:mydb.db?vfs=litestream
SELECT * FROM my_table;

import os
import sqlite3

os.environ["LITESTREAM_REPLICA_URL"] = "s3://my-bucket/mydb"

conn = sqlite3.connect(":memory:")
conn.enable_load_extension(True)
conn.load_extension("./dist/litestream-vfs.so")

# Open database using the litestream VFS
conn = sqlite3.connect("file:mydb.db?vfs=litestream")
cursor = conn.execute("SELECT * FROM users")
for row in cursor:
    print(row)

import (
    "database/sql"
    "os"

    _ "github.com/mattn/go-sqlite3"
)

func main() {
    os.Setenv("LITESTREAM_REPLICA_URL", "s3://my-bucket/mydb")

    db, err := sql.Open("sqlite3", "file:mydb.db?vfs=litestream")
    if err != nil {
        panic(err)
    }
    defer db.Close()

    rows, err := db.Query("SELECT * FROM users")
    // ...
}

SELECT litestream_txid();
-- Returns: "0000000000000042"

SELECT litestream_time();
-- Returns: "2024-01-15T10:30:00.123456789Z" or "latest"

SELECT litestream_lag();
-- Returns: 2 (seconds behind primary)

-- View database as of a specific timestamp (RFC3339 format)
SELECT litestream_set_time('2024-01-15T10:30:00Z');

-- Relative time expressions are also supported
SELECT litestream_set_time('5 minutes ago');
SELECT litestream_set_time('yesterday');
SELECT litestream_set_time('2 hours ago');

-- Return to latest state
SELECT litestream_set_time('LATEST');

-- Check current count
SELECT COUNT(*) FROM orders;
-- Returns: 1000

-- Go back in time
SELECT litestream_set_time('1 hour ago');

-- Check historical count
SELECT COUNT(*) FROM orders;
-- Returns: 950

-- Return to present
SELECT litestream_set_time('LATEST');

export LITESTREAM_REPLICA_URL="s3://my-bucket/mydb"
export LITESTREAM_WRITE_ENABLED="true"
export LITESTREAM_SYNC_INTERVAL="1s"

.load ./dist/litestream-vfs.so
.open file:newdb.db?vfs=litestream

CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);
INSERT INTO users (name) VALUES ('Alice');
-- Data is synced to remote storage automatically

export AWS_ACCESS_KEY_ID="your-key"
export AWS_SECRET_ACCESS_KEY="your-secret"
export AWS_REGION="us-east-1"

# For S3-compatible services:
export LITESTREAM_REPLICA_URL="s3://bucket/path?endpoint=https://custom.endpoint.com"

export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"
export LITESTREAM_REPLICA_URL="gs://bucket/path"

export LITESTREAM_LOG_LEVEL="DEBUG"