litestream

package module
v0.5.6-0...-c09d3c3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 5, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

README

Litestream for the ncruces/go-sqlite3 driver

This is a fork focusing on library use, which uses the ncruces/go-sqlite3 driver.

If you can use the litestream CLI for streaming replication, please do. The ncruces/go-sqlite3 driver is compatible with the CLI on all major platforms.

Otherwise, check the examples for how to use.

Lightweight read-replicas

This also implements the "litestream" SQLite VFS that offers Litestream lightweight read-replicas for the ncruces/go-sqlite3 driver.

Our VFS API has significant differences from upstream; follow the usage example.

Our PRAGMA litestream_time accepts:

Documentation

Index

Constants

View Source 
const (
	DefaultMonitorInterval      = 1 * time.Second
	DefaultCheckpointInterval   = 1 * time.Minute
	DefaultBusyTimeout          = 1 * time.Second
	DefaultMinCheckpointPageN   = 1000
	DefaultTruncatePageN        = 121359 // ~500MB with 4KB page size
	DefaultShutdownSyncTimeout  = 30 * time.Second
	DefaultShutdownSyncInterval = 500 * time.Millisecond
)

Default DB settings.

View Source 
const (
	CheckpointModePassive  = "PASSIVE"
	CheckpointModeFull     = "FULL"
	CheckpointModeRestart  = "RESTART"
	CheckpointModeTruncate = "TRUNCATE"
)

SQLite checkpoint modes.

View Source 
const (
	WALHeaderChecksumOffset      = 24
	WALFrameHeaderChecksumOffset = 16
)

SQLite WAL constants.

View Source 
const (
	// WALHeaderSize is the size of the WAL header, in bytes.
	WALHeaderSize = 32

	// WALFrameHeaderSize is the size of the WAL frame header, in bytes.
	WALFrameHeaderSize = 24
)
View Source 
const (
	// The default poll interval.
	DefaultPollInterval = 1 * time.Second

	// The default cache size: 10 MiB.
	DefaultCacheSize = 10 * 1024 * 1024
)
View Source 
const (
	DefaultSnapshotInterval  = 24 * time.Hour
	DefaultSnapshotRetention = 24 * time.Hour

	DefaultRetention              = 24 * time.Hour
	DefaultRetentionCheckInterval = 1 * time.Hour

	// DefaultL0Retention is the default time that L0 files are kept around
	// after they have been compacted into L1 files.
	DefaultL0Retention = 5 * time.Minute
	// DefaultL0RetentionCheckInterval controls how frequently L0 retention is
	// enforced. This interval should be more frequent than the L1 compaction
	// interval so that VFS read replicas have time to observe new files.
	DefaultL0RetentionCheckInterval = 15 * time.Second
)

Store defaults

View Source 
const DefaultEstimatedPageIndexSize = 32 * 1024 // 32KB

DefaultEstimatedPageIndexSize is size that is first fetched when fetching the page index. If the fetch was smaller than the actual page index, another call is made to fetch the rest.

View Source 
const DefaultRestoreParallelism = 8

DefaultRestoreParallelism is the default parallelism when downloading WAL files.

View Source 
const (
	DefaultSyncInterval = 1 * time.Second
)

Default replica settings.

View Source 
const (
	MetaDirSuffix = "-litestream"
)

Naming constants.

View Source 
const SnapshotLevel = 9

SnapshotLevel represents the level which full snapshots are held.

Variables

View Source 
var (
	ErrNoSnapshots      = errors.New("no snapshots available")
	ErrChecksumMismatch = errors.New("invalid replica, checksum mismatch")
)

Litestream errors.

View Source 
var (
	// LogWriter is the destination writer for all logging.
	LogWriter = os.Stdout

	// LogFlags are the flags passed to log.New().
	LogFlags = 0
)
View Source 
var (
	// ErrNoCompaction is returned when no new files are available from the previous level.
	ErrNoCompaction = errors.New("no compaction")

	// ErrCompactionTooEarly is returned when a compaction is attempted too soon
	// since the last compaction time. This is used to prevent frequent
	// re-compaction when restarting the process.
	ErrCompactionTooEarly = errors.New("compaction too early")

	// ErrTxNotAvailable is returned when a transaction does not exist.
	ErrTxNotAvailable = errors.New("transaction not available")

	// ErrDBNotReady is returned when compaction is attempted before the
	// database has been initialized (e.g., page size not yet known).
	ErrDBNotReady = errors.New("db not ready")
)
View Source 
var ErrStopIter = errors.New("stop iterator")

Functions

func BoolQueryValue 

func BoolQueryValue(query url.Values, keys ...string) (value bool, ok bool)

BoolQueryValue returns a boolean value from URL query parameters. It checks multiple keys in order and returns the value and whether it was set.

func CalcRestorePlan 

func CalcRestorePlan(ctx context.Context, client ReplicaClient, txID ltx.TXID, timestamp time.Time, logger *slog.Logger) ([]*ltx.FileInfo, error)

CalcRestorePlan returns a list of storage paths to restore a snapshot at the given TXID.

func Checksum 

func Checksum(bo binary.ByteOrder, s0, s1 uint32, b []byte) (uint32, uint32)

Checksum computes a running SQLite checksum over a byte slice.

func CleanReplicaURLPath 

func CleanReplicaURLPath(p string) string

CleanReplicaURLPath cleans a URL path for use in replica storage.

func CreateVFSReadReplica 

func CreateVFSReadReplica(name string, client ReplicaClient, options VFSOptions)

CreateVFSReadReplica creates a read-replica VFS from a client.

func FetchLTXHeader 

func FetchLTXHeader(ctx context.Context, client ReplicaClient, info *ltx.FileInfo) (ltx.Header, error)

FetchLTXHeader reads & returns the LTX header for the given file info.

func FetchPage 

func FetchPage(ctx context.Context, client ReplicaClient, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (ltx.PageHeader, []byte, error)

FetchPage fetches and decodes a single page frame from an LTX file.

func FetchPageIndex 

func FetchPageIndex(ctx context.Context, client ReplicaClient, info *ltx.FileInfo) (map[uint32]ltx.PageIndexElem, error)

func FindLTXFiles 

func FindLTXFiles(ctx context.Context, client ReplicaClient, level int, useMetadata bool, filter func(*ltx.FileInfo) (bool, error)) ([]*ltx.FileInfo, error)

FindLTXFiles returns a list of files that match filter. The useMetadata parameter is passed through to LTXFiles to control whether accurate timestamps are fetched from metadata. When true (timestamp-based restore), accurate timestamps are required. When false (normal operations), fast timestamps are sufficient.

func IsBackblazeEndpoint 

func IsBackblazeEndpoint(endpoint string) bool

IsBackblazeEndpoint returns true if the endpoint is Backblaze B2.

func IsCloudflareR2Endpoint 

func IsCloudflareR2Endpoint(endpoint string) bool

IsCloudflareR2Endpoint returns true if the endpoint is Cloudflare R2.

func IsDigitalOceanEndpoint 

func IsDigitalOceanEndpoint(endpoint string) bool

IsDigitalOceanEndpoint returns true if the endpoint is Digital Ocean Spaces.

func IsFilebaseEndpoint 

func IsFilebaseEndpoint(endpoint string) bool

IsFilebaseEndpoint returns true if the endpoint is Filebase.

func IsMinIOEndpoint 

func IsMinIOEndpoint(endpoint string) bool

IsMinIOEndpoint returns true if the endpoint appears to be MinIO or similar (a custom endpoint with a port number that is not a known cloud provider).

func IsScalewayEndpoint 

func IsScalewayEndpoint(endpoint string) bool

IsScalewayEndpoint returns true if the endpoint is Scaleway Object Storage.

func IsTigrisEndpoint 

func IsTigrisEndpoint(endpoint string) bool

IsTigrisEndpoint returns true if the endpoint is the Tigris object storage service.

func IsURL 

func IsURL(s string) bool

func LTXDir 

func LTXDir(root string) string

LTXDir returns the path to an LTX directory.

func LTXFilePath 

func LTXFilePath(root string, level int, minTXID, maxTXID ltx.TXID) string

LTXFilePath returns the path to a single LTX file.

func LTXLevelDir 

func LTXLevelDir(root string, level int) string

LTXLevelDir returns the path to an LTX level directory.

func ParseReplicaURL 

func ParseReplicaURL(s string) (scheme, host, urlPath string, err error)

ParseReplicaURL parses a replica URL and returns the scheme, host, and path.

func ParseReplicaURLWithQuery 

func ParseReplicaURLWithQuery(s string) (scheme, host, urlPath string, query url.Values, userinfo *url.Userinfo, err error)

ParseReplicaURLWithQuery parses a replica URL and returns query parameters and userinfo.

func RegionFromS3ARN 

func RegionFromS3ARN(arn string) string

RegionFromS3ARN extracts the region from an S3 ARN.

func RegisterReplicaClientFactory 

func RegisterReplicaClientFactory(scheme string, factory ReplicaClientFactory)

RegisterReplicaClientFactory registers a factory function for creating replica clients for a given URL scheme. This is typically called from init() functions in backend packages.

func RemoveVFSReadReplica 

func RemoveVFSReadReplica(name string)

RemoveVFSReadReplica removes a Litestream read-replica VFS by name.

func ReplicaTypeFromURL 

func ReplicaTypeFromURL(rawURL string) string

ReplicaTypeFromURL returns the replica type from a URL string. Returns empty string if the URL is invalid or has no scheme.

func WALChecksum 

func WALChecksum(bo binary.ByteOrder, s0, s1 uint32, b []byte) (uint32, uint32)

WALChecksum computes a running SQLite WAL checksum over a byte slice.

Types

type CompactionLevel 

type CompactionLevel struct {
	// The numeric level. Must match the index in the list of levels.
	Level int

	// The frequency that the level is compacted from the previous level.
	Interval time.Duration
}

CompactionLevel represents a single part of a multi-level compaction. Each level merges LTX files from the previous level into larger time granularities.

func (*CompactionLevel) NextCompactionAt 

func (lvl *CompactionLevel) NextCompactionAt(now time.Time) time.Time

NextCompactionAt returns the time until the next compaction occurs. Returns the current time if it is exactly a multiple of the level interval.

func (*CompactionLevel) PrevCompactionAt 

func (lvl *CompactionLevel) PrevCompactionAt(now time.Time) time.Time

PrevCompactionAt returns the time when the last compaction occurred. Returns the current time if it is exactly a multiple of the level interval.

type CompactionLevels 

type CompactionLevels []*CompactionLevel

CompactionLevels represents a sorted slice of non-snapshot compaction levels.

func (CompactionLevels) IsValidLevel 

func (a CompactionLevels) IsValidLevel(level int) bool

IsValidLevel returns true if level is a valid compaction level number.

func (CompactionLevels) Level 

func (a CompactionLevels) Level(level int) (*CompactionLevel, error)

Level returns the compaction level at the given index. Returns an error if the index is a snapshot level or is out of bounds.

func (CompactionLevels) MaxLevel 

func (a CompactionLevels) MaxLevel() int

MaxLevel return the highest non-snapshot compaction level.

func (CompactionLevels) NextLevel 

func (a CompactionLevels) NextLevel(level int) int

NextLevel returns the next compaction level. Returns -1 if there is no next level.

func (CompactionLevels) PrevLevel 

func (a CompactionLevels) PrevLevel(level int) int

PrevLevel returns the previous compaction level. Returns -1 if there is no previous level.

func (CompactionLevels) Validate 

func (a CompactionLevels) Validate() error

Validate returns an error if the levels are invalid.

type DB 

type DB struct {

	// Minimum threshold of WAL size, in pages, before a passive checkpoint.
	// A passive checkpoint will attempt a checkpoint but fail if there are
	// active transactions occurring at the same time.
	//
	// Uses PASSIVE checkpoint mode (non-blocking). Keeps WAL size manageable
	// for faster restores. Default: 1000 pages (~4MB with 4KB page size).
	MinCheckpointPageN 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.
	//
	// Uses TRUNCATE checkpoint mode (blocking). Prevents unbounded WAL growth
	// from long-lived read transactions. Default: 121359 pages (~500MB with 4KB
	// page size). Set to 0 to disable forced truncation (use with caution as
	// WAL can grow unbounded if read transactions prevent checkpointing).
	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.
	//
	// Uses PASSIVE checkpoint mode (non-blocking). Default: 1 minute.
	// Set to 0 to disable time-based checkpoints.
	CheckpointInterval time.Duration

	// Frequency at which to perform db sync.
	MonitorInterval time.Duration

	// The timeout to wait for EBUSY from SQLite.
	BusyTimeout time.Duration

	// Minimum time to retain L0 files after they have been compacted into L1.
	L0Retention time.Duration

	// Remote replica for the database.
	// Must be set before calling Open().
	Replica *Replica

	// Shutdown sync retry settings.
	// ShutdownSyncTimeout is the total time to retry syncing on shutdown.
	// ShutdownSyncInterval is the time between retry attempts.
	ShutdownSyncTimeout  time.Duration
	ShutdownSyncInterval time.Duration

	// Where to send log messages, defaults to global slog with database epath.
	Logger *slog.Logger
	// contains filtered or unexported fields
}

DB represents a managed instance of a SQLite database in the file system.

Checkpoint Strategy: Litestream uses a progressive 3-tier checkpoint approach to balance WAL size management with write availability:

  1. MinCheckpointPageN (PASSIVE): Non-blocking checkpoint at ~1k pages (~4MB). Attempts checkpoint but allows concurrent readers/writers.

  2. CheckpointInterval (PASSIVE): Time-based non-blocking checkpoint. Ensures regular checkpointing even with low write volume.

  3. TruncatePageN (TRUNCATE): Blocking checkpoint at ~121k pages (~500MB). Emergency brake for runaway WAL growth. Can block writes while waiting for long-lived read transactions. Configurable/disableable.

The RESTART checkpoint mode was permanently removed due to production issues with indefinite write blocking (issue #724). All checkpoints now use either PASSIVE (non-blocking) or TRUNCATE (emergency only) modes.

func NewDB 

func NewDB(path string) *DB

NewDB returns a new instance of DB for a given path.

func (*DB) CRC64 

func (db *DB) CRC64(ctx context.Context) (uint64, ltx.Pos, error)

CRC64 returns a CRC-64 ISO checksum of the database and its current position.

This function obtains a read lock so it prevents syncs from occurring until the operation is complete. The database will still be usable but it will be unable to checkpoint during this time.

If dst is set, the database file is copied to that location before checksum.

func (*DB) Checkpoint 

func (db *DB) Checkpoint(ctx context.Context, mode string) (err error)

Checkpoint performs a checkpoint on the WAL file.

func (*DB) Close 

func (db *DB) Close(ctx context.Context) (err error)

Close flushes outstanding WAL writes to replicas, releases the read lock, and closes the database. Takes a context for final sync.

func (*DB) Compact 

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

Compact performs a compaction of the LTX file at the previous level into dstLevel. Returns metadata for the newly written compaction file. Returns ErrNoCompaction if no new files are available to be compacted.

func (*DB) DirInfo 

func (db *DB) DirInfo() os.FileInfo

DirInfo returns the cached file stats for the parent directory of the database file when it was initialized.

func (*DB) EnforceL0RetentionByTime 

func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error

EnforceL0RetentionByTime retains L0 files until they have been compacted into L1 and have existed for at least L0Retention.

func (*DB) EnforceRetentionByTXID 

func (db *DB) EnforceRetentionByTXID(ctx context.Context, level int, txID ltx.TXID) (err error)

EnforceRetentionByTXID enforces retention so that any LTX files below the target TXID are deleted. Always keep at least one file.

func (*DB) EnforceSnapshotRetention 

func (db *DB) EnforceSnapshotRetention(ctx context.Context, timestamp time.Time) (minSnapshotTXID ltx.TXID, err error)

EnforceSnapshotRetention enforces retention of the snapshot level in the database by timestamp.

func (*DB) FileInfo 

func (db *DB) FileInfo() os.FileInfo

FileInfo returns the cached file stats for the database file when it was initialized.

func (*DB) LTXDir 

func (db *DB) LTXDir() string

LTXDir returns path of the root LTX directory.

func (*DB) LTXLevelDir 

func (db *DB) LTXLevelDir(level int) string

LTXLevelDir returns path of the given LTX compaction level. Panics if level is negative.

func (*DB) LTXPath 

func (db *DB) LTXPath(level int, minTXID, maxTXID ltx.TXID) string

LTXPath returns the local path of a single LTX file. Panics if level or either txn ID is negative.

func (*DB) MaxLTX 

func (db *DB) MaxLTX() (minTXID, maxTXID ltx.TXID, err error)

MaxLTX returns the last LTX file written to level 0.

func (*DB) MaxLTXFileInfo 

func (db *DB) MaxLTXFileInfo(ctx context.Context, level int) (ltx.FileInfo, error)

MaxLTXFileInfo returns the metadata for the last LTX file in a level. If cached, it will returned the local copy. Otherwise, it fetches from the replica.

func (*DB) MetaPath 

func (db *DB) MetaPath() string

MetaPath returns the path to the database metadata.

func (*DB) Notify 

func (db *DB) Notify() <-chan struct{}

Notify returns a channel that closes when the shadow WAL changes.

func (*DB) Open 

func (db *DB) Open() (err error)

Open initializes the background monitoring goroutine.

func (*DB) PageSize 

func (db *DB) PageSize() int

PageSize returns the page size of the underlying database. Only valid after database exists & Init() has successfully run.

func (*DB) Path 

func (db *DB) Path() string

Path returns the path to the database.

func (*DB) Pos 

func (db *DB) Pos() (ltx.Pos, error)

Pos returns the current replication position of the database.

func (*DB) SQLDB 

func (db *DB) SQLDB() *sql.DB

SQLDB returns a reference to the underlying sql.DB connection.

func (*DB) SetMetaPath 

func (db *DB) SetMetaPath(path string)

SetMetaPath sets the path to database metadata.

func (*DB) Snapshot 

func (db *DB) Snapshot(ctx context.Context) (*ltx.FileInfo, error)

SnapshotDB writes a snapshot to the replica for the current position of the database.

func (*DB) SnapshotReader 

func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error)

SnapshotReader returns the current position of the database & a reader that contains a full database snapshot.

func (*DB) Sync 

func (db *DB) Sync(ctx context.Context) (err error)

Sync copies pending data from the WAL to the shadow WAL.

func (*DB) WALPath 

func (db *DB) WALPath() string

WALPath returns the path to the database's WAL file.

type PrevFrameMismatchError 

type PrevFrameMismatchError struct {
	Err error
}

func (*PrevFrameMismatchError) Error 

func (e *PrevFrameMismatchError) Error() string

func (*PrevFrameMismatchError) Unwrap 

func (e *PrevFrameMismatchError) Unwrap() error

type Replica 

type Replica struct {

	// Client used to connect to the remote replica.
	Client ReplicaClient

	// Time between syncs with the shadow WAL.
	SyncInterval time.Duration

	// If true, replica monitors database for changes automatically.
	// Set to false if replica is being used synchronously (such as in tests).
	MonitorEnabled bool
	// contains filtered or unexported fields
}

Replica connects a database to a replication destination via a ReplicaClient. The replica manages periodic synchronization and maintaining the current replica position.

func NewReplica 

func NewReplica(db *DB) *Replica

func NewReplicaWithClient 

func NewReplicaWithClient(db *DB, client ReplicaClient) *Replica

func (*Replica) CalcRestoreTarget 

func (r *Replica) CalcRestoreTarget(ctx context.Context, opt RestoreOptions) (updatedAt time.Time, err error)

CalcRestoreTarget returns a target time restore from.

func (*Replica) CreatedAt 

func (r *Replica) CreatedAt(ctx context.Context) (time.Time, error)

CreatedAt returns the earliest creation time of any LTX file. Returns zero time if no LTX files exist.

func (*Replica) DB 

func (r *Replica) DB() *DB

DB returns a reference to the database the replica is attached to, if any.

func (*Replica) EnforceRetention 

func (r *Replica) EnforceRetention(ctx context.Context) (err error)

EnforceRetention forces a new snapshot once the retention interval has passed. Older snapshots and WAL files are then removed.

func (*Replica) Logger 

func (r *Replica) Logger() *slog.Logger

Logger returns the DB sub-logger for this replica.

func (*Replica) MaxLTXFileInfo 

func (r *Replica) MaxLTXFileInfo(ctx context.Context, level int) (info ltx.FileInfo, err error)

MaxLTXFileInfo returns metadata about the last LTX file for a given level. Returns nil if no files exist for the level.

func (*Replica) Pos 

func (r *Replica) Pos() ltx.Pos

Pos returns the current replicated position. Returns a zero value if the current position cannot be determined.

func (*Replica) Restore 

func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error)

Replica restores the database from a replica based on the options given. This method will restore into opt.OutputPath, if specified, or into the DB's original database path. It can optionally restore from a specific replica or it will automatically choose the best one. Finally, a timestamp can be specified to restore the database to a specific point-in-time.

func (*Replica) SetPos 

func (r *Replica) SetPos(pos ltx.Pos)

SetPos sets the current replicated position.

func (*Replica) Start 

func (r *Replica) Start(ctx context.Context) error

Starts replicating in a background goroutine.

func (*Replica) Stop 

func (r *Replica) Stop(hard bool) (err error)

Stop cancels any outstanding replication and blocks until finished.

Performing a hard stop will close the DB file descriptor which could release locks on per-process locks. Hard stops should only be performed when stopping the entire process.

func (*Replica) Sync 

func (r *Replica) Sync(ctx context.Context) (err error)

Sync copies new WAL frames from the shadow WAL to the replica client. Only one Sync can run at a time to prevent concurrent uploads of the same file.

func (*Replica) TimeBounds 

func (r *Replica) TimeBounds(ctx context.Context) (createdAt, updatedAt time.Time, err error)

TimeBounds returns the creation time & last updated time. Returns zero time if LTX files exist.

type ReplicaClient 

type ReplicaClient interface {
	// Type returns the type of client.
	Type() string

	// Init initializes the replica client connection.
	// This may establish connections, validate configuration, etc.
	// Implementations should be idempotent (no-op if already initialized).
	Init(ctx context.Context) error

	// LTXFiles returns an iterator of all LTX files on the replica for a given level.
	// If seek is specified, the iterator start from the given TXID or the next available if not found.
	// If useMetadata is true, the iterator fetches accurate timestamps from metadata for timestamp-based restore.
	// When false, the iterator uses fast timestamps (LastModified/Created/ModTime) for normal operations.
	LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)

	// OpenLTXFile returns a reader that contains an LTX file at a given TXID.
	// If seek is specified, the reader will start at the given offset.
	// Returns an os.ErrNotFound error if the LTX file does not exist.
	OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)

	// WriteLTXFile writes an LTX file to the replica.
	// Returns metadata for the written file.
	WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error)

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

	// DeleteAll deletes all files.
	DeleteAll(ctx context.Context) error
}

ReplicaClient represents client to connect to a Replica.

func NewReplicaClientFromURL 

func NewReplicaClientFromURL(rawURL string) (ReplicaClient, error)

NewReplicaClientFromURL creates a new ReplicaClient from a URL string. The URL scheme determines which backend is used (s3, gs, abs, file, etc.).

type ReplicaClientFactory 

type ReplicaClientFactory func(scheme, host, urlPath string, query url.Values, userinfo *url.Userinfo) (ReplicaClient, error)

ReplicaClientFactory is a function that creates a ReplicaClient from URL components. The userinfo parameter contains credentials from the URL (e.g., user:pass@host).

type RestoreOptions 

type RestoreOptions struct {
	// Target path to restore into.
	// If blank, the original DB path is used.
	OutputPath string

	// Specific transaction to restore to.
	// If zero, TXID is ignored.
	TXID ltx.TXID

	// Point-in-time to restore database.
	// If zero, database restore to most recent state available.
	Timestamp time.Time

	// Specifies how many WAL files are downloaded in parallel during restore.
	Parallelism int
}

RestoreOptions represents options for DB.Restore().

func NewRestoreOptions 

func NewRestoreOptions() RestoreOptions

NewRestoreOptions returns a new instance of RestoreOptions with defaults.

type Store 

type Store struct {

	// The frequency of snapshots.
	SnapshotInterval time.Duration
	// The duration of time that snapshots are kept before being deleted.
	SnapshotRetention time.Duration

	// The duration that L0 files are kept after being compacted into L1.
	L0Retention time.Duration
	// How often to check for expired L0 files.
	L0RetentionCheckInterval time.Duration

	// If true, compaction is run in the background according to compaction levels.
	CompactionMonitorEnabled bool

	// Shutdown sync retry settings.
	ShutdownSyncTimeout  time.Duration
	ShutdownSyncInterval time.Duration
	// contains filtered or unexported fields
}

Store represents the top-level container for databases.

It manages async background tasks like compactions so that the system is not overloaded by too many concurrent tasks.

func NewStore 

func NewStore(dbs []*DB, levels CompactionLevels) *Store

func (*Store) AddDB 

func (s *Store) AddDB(db *DB) error

AddDB registers a new database with the store and starts monitoring it.

func (*Store) Close 

func (s *Store) Close(ctx context.Context) (err error)

func (*Store) CompactDB 

func (s *Store) CompactDB(ctx context.Context, db *DB, lvl *CompactionLevel) (*ltx.FileInfo, error)

CompactDB performs a compaction or snapshot for a given database on a single destination level. This function will only proceed if a compaction has not occurred before the last compaction time.

func (*Store) DBs 

func (s *Store) DBs() []*DB

func (*Store) EnforceSnapshotRetention 

func (s *Store) EnforceSnapshotRetention(ctx context.Context, db *DB) error

EnforceSnapshotRetention removes old snapshots by timestamp and then cleans up all lower levels based on minimum snapshot TXID.

func (*Store) Open 

func (s *Store) Open(ctx context.Context) error

func (*Store) RemoveDB 

func (s *Store) RemoveDB(ctx context.Context, path string) error

RemoveDB stops monitoring the database at the provided path and closes it.

func (*Store) SetL0Retention 

func (s *Store) SetL0Retention(d time.Duration)

SetL0Retention updates the retention window for L0 files and propagates it to all managed databases.

func (*Store) SetShutdownSyncInterval 

func (s *Store) SetShutdownSyncInterval(d time.Duration)

SetShutdownSyncInterval updates the shutdown sync interval and propagates it to all managed databases.

func (*Store) SetShutdownSyncTimeout 

func (s *Store) SetShutdownSyncTimeout(d time.Duration)

SetShutdownSyncTimeout updates the shutdown sync timeout and propagates it to all managed databases.

func (*Store) SnapshotLevel 

func (s *Store) SnapshotLevel() *CompactionLevel

SnapshotLevel returns a pseudo compaction level based on snapshot settings.

type VFSOptions 

type VFSOptions struct {
	// Where to log error messages. May be nil.
	Logger *slog.Logger

	// Replica poll interval.
	// Should be less than the compaction interval
	// used by the [ReplicaClient] at MinLevel+1.
	PollInterval time.Duration

	// CacheSize is the maximum size for the page cache in bytes.
	// Zero means [DefaultCacheSize], negative disables caching.
	CacheSize int
}

VFSOptions represents options for CreateVFSReadReplica.

type WALReader 

type WALReader struct {
	// contains filtered or unexported fields
}

WALReader wraps an io.Reader and parses SQLite WAL frames.

This reader verifies the salt & checksum integrity while it reads. It does not enforce transaction boundaries (i.e. it may return uncommitted frames). It is the responsibility of the caller to handle this.

func NewWALReader 

func NewWALReader(rd io.ReaderAt, logger *slog.Logger) (*WALReader, error)

NewWALReader returns a new instance of WALReader.

func NewWALReaderWithOffset 

func NewWALReaderWithOffset(ctx context.Context, rd io.ReaderAt, offset int64, salt1, salt2 uint32, logger *slog.Logger) (*WALReader, error)

NewWALReaderWithOffset returns a new instance of WALReader at a given offset. Salt must match or else no frames will be returned. Checksum calculated from from previous page.

func (*WALReader) FrameSaltsUntil 

func (r *WALReader) FrameSaltsUntil(ctx context.Context, until [2]uint32) (map[[2]uint32]struct{}, error)

FrameSaltsUntil returns a set of all unique frame salts in the WAL file.

func (*WALReader) Offset 

func (r *WALReader) Offset() int64

Offset returns the file offset of the last read frame. Returns zero if no frames have been read.

func (*WALReader) PageMap 

func (r *WALReader) PageMap(ctx context.Context) (m map[uint32]int64, maxOffset int64, commit uint32, err error)

PageMap reads all committed frames until the end of the file and returns a map of pgno to offset of the latest version of each page. Also returns the max offset of the wal segment read, and the final database size, in pages.

func (*WALReader) PageSize 

func (r *WALReader) PageSize() uint32

PageSize returns the page size from the header. Must call ReadHeader() first.

func (*WALReader) ReadFrame 

func (r *WALReader) ReadFrame(ctx context.Context, data []byte) (pgno, commit uint32, err error)

ReadFrame reads the next frame from the WAL and returns the page number. Returns io.EOF at the end of the valid WAL.

Source Files

View all Source files

Directories

Path Synopsis
_examples
library/basic command
Example: Basic Litestream Library Usage
 Example: Basic Litestream Library Usage
library/s3 command
Example: Litestream Library Usage with S3 and Restore-on-Startup
 Example: Litestream Library Usage with S3 and Restore-on-Startup
library/vfs command
Example: Basic Litestream Library Usage
 Example: Basic Litestream Library Usage
testingutil

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.