Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

replicate

package module

Go to main page

Versions in this module

v0

v0.8.0

Jun 7, 2026

Changes in this version

+ func ShardedKey(rootSegment, symbol, suffix string) string

+ func ShardedPrefix(symbol string) string

+ type SyncPolicy uint8

+ const SyncAll

+ const SyncLocalOnly

+ const SyncQuorumW2

+ func (p SyncPolicy) String() string

v0.7.3

Jun 7, 2026

Changes in this version

type Leaser

+ MaxLeaseTTL func() time.Duration

v0.7.2

May 13, 2026

v0.7.1

Apr 27, 2026

v0.7.0

Apr 10, 2026

v0.6.0

Apr 10, 2026

v0.5.13

Apr 27, 2026

v0.5.12

Apr 10, 2026

v0.4.0

Apr 10, 2026

Changes in this version

+ const CheckpointModeFull

+ const CheckpointModePassive

+ const CheckpointModeRestart

+ const CheckpointModeTruncate

+ const DefaultBusyTimeout

+ const DefaultCheckpointInterval

+ const DefaultDBInitTimeout

+ const DefaultEstimatedPageIndexSize

+ const DefaultFollowInterval

+ const DefaultHeartbeatCheckInterval

+ const DefaultHeartbeatInterval

+ const DefaultHeartbeatTimeout

+ const DefaultL0Retention

+ const DefaultL0RetentionCheckInterval

+ const DefaultMinCheckpointPageN

+ const DefaultMonitorInterval

+ const DefaultRestoreParallelism

+ const DefaultRetention

+ const DefaultRetentionCheckInterval

+ const DefaultShutdownSyncInterval

+ const DefaultShutdownSyncTimeout

+ const DefaultSnapshotInterval

+ const DefaultSnapshotRetention

+ const DefaultSyncBackoffMax

+ const DefaultSyncInterval

+ const DefaultTruncatePageN

+ const GenerationsDirV3

+ const LogKeyDB

+ const LogKeySubsystem

+ const LogKeySystem

+ const LogSubsystemCompactor

+ const LogSubsystemWALReader

+ const LogSystemServer

+ const LogSystemStore

+ const MetaDirSuffix

+ const MinHeartbeatInterval

+ const SnapshotLevel

+ const SnapshotsDirV3

+ const SyncErrorLogInterval

+ const WALDirV3

+ const WALFrameHeaderChecksumOffset

+ const WALFrameHeaderSize

+ const WALHeaderChecksumOffset

+ const WALHeaderSize

+ var DefaultCompactionLevels = CompactionLevels

+ var ErrChecksumMismatch = errors.New("invalid replica, checksum mismatch")

+ var ErrCompactionTooEarly = errors.New("compaction too early")

+ var ErrDBNotReady = &DBNotReadyError

+ var ErrDatabaseNotFound = errors.New("database not found")

+ var ErrDatabaseNotOpen = errors.New("database not open")

+ var ErrLTXCorrupted = errors.New("ltx file corrupted")

+ var ErrLTXMissing = errors.New("ltx file missing")

+ var ErrLeaseNotHeld = errors.New("lease not held")

+ var ErrNoCompaction = errors.New("no compaction")

+ var ErrNoSnapshots = errors.New("no snapshots available")

+ var ErrShutdownInterrupted = errors.New("shutdown sync interrupted")

+ var ErrStopIter = errors.New("stop iterator")

+ var ErrTxNotAvailable = errors.New("transaction not available")

+ var LogFlags = 0

+ var LogWriter = os.Stdout

+ func AutoReplicate(dbPath string) func()

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

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

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

+ func CleanReplicaURLPath(p string) string

+ func EnsureEndpointScheme(endpoint string) (string, bool)

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

+ func FetchPage(ctx context.Context, client ReplicaClient, level int, ...) (ltx.PageHeader, []byte, error)

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

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

+ func FormatSnapshotFilenameV3(index int) string

+ func FormatWALSegmentFilenameV3(index int, offset int64) string

+ func GenerationPathV3(root, generation string) string

+ func GenerationsPathV3(root string) string

+ func IsBackblazeEndpoint(endpoint string) bool

+ func IsCloudflareR2Endpoint(endpoint string) bool

+ func IsDigitalOceanEndpoint(endpoint string) bool

+ func IsFilebaseEndpoint(endpoint string) bool

+ func IsGenerationIDV3(s string) bool

+ func IsHetznerEndpoint(endpoint string) bool

+ func IsLocalEndpoint(endpoint string) bool

+ func IsMinIOEndpoint(endpoint string) bool

+ func IsScalewayEndpoint(endpoint string) bool

+ func IsSupabaseEndpoint(endpoint string) bool

+ func IsTigrisEndpoint(endpoint string) bool

+ func IsURL(s string) bool

+ func LTXDir(root string) string

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

+ func LTXLevelDir(root string, level int) string

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

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

+ func ParseSnapshotFilenameV3(filename string) (index int, err error)

+ func ParseWALSegmentFilenameV3(filename string) (index int, offset int64, err error)

+ func ReadTXIDFile(outputPath string) (ltx.TXID, error)

+ func RegionFromS3ARN(arn string) string

+ func RegisterReplicaClientFactory(scheme string, factory ReplicaClientFactory)

+ func ReplicaTypeFromURL(rawURL string) string

+ func SnapshotPathV3(root, generation string, index int) string

+ func SnapshotsPathV3(root, generation string) string

+ func TXIDPath(outputPath string) string

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

+ func WALPathV3(root, generation string) string

+ func WALSegmentPathV3(root, generation string, index int, offset int64) string

+ func WriteTXIDFile(outputPath string, txid ltx.TXID) error

+ type CompactionLevel struct

+ Interval time.Duration

+ Level int

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

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

+ type CompactionLevels []*CompactionLevel

+ func (a CompactionLevels) IsValidLevel(level int) bool

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

+ func (a CompactionLevels) MaxLevel() int

+ func (a CompactionLevels) NextLevel(level int) int

+ func (a CompactionLevels) PrevLevel(level int) int

+ func (a CompactionLevels) Validate() error

+ type Compactor struct

+ AgeIdentities []age.Identity

+ AgeRecipients []age.Recipient

+ CacheGetter func(level int) (*ltx.FileInfo, bool)

+ CacheSetter func(level int, info *ltx.FileInfo)

+ CompactionVerifyErrorCounter prometheus.Counter

+ LocalFileDeleter func(level int, minTXID, maxTXID ltx.TXID) error

+ LocalFileOpener func(level int, minTXID, maxTXID ltx.TXID) (io.ReadCloser, error)

+ RetentionEnabled bool

+ VerifyCompaction bool

+ func NewCompactor(client ReplicaClient, logger *slog.Logger) *Compactor

+ func (c *Compactor) Compact(ctx context.Context, dstLevel int) (*ltx.FileInfo, error)

+ func (c *Compactor) EnforceL0Retention(ctx context.Context, retention time.Duration) error

+ func (c *Compactor) EnforceRetentionByTXID(ctx context.Context, level int, txID ltx.TXID) error

+ func (c *Compactor) EnforceSnapshotRetention(ctx context.Context, retention time.Duration) (ltx.TXID, error)

+ func (c *Compactor) MaxLTXFileInfo(ctx context.Context, level int) (ltx.FileInfo, error)

+ func (c *Compactor) VerifyLevelConsistency(ctx context.Context, level int) error

+ type DB struct

+ BusyTimeout time.Duration

+ CheckpointInterval time.Duration

+ Done <-chan struct{}

+ L0Retention time.Duration

+ Logger *slog.Logger

+ MinCheckpointPageN int

+ MonitorInterval time.Duration

+ Replica *Replica

+ RetentionEnabled bool

+ ShutdownSyncInterval time.Duration

+ ShutdownSyncTimeout time.Duration

+ TruncatePageN int

+ VerifyCompaction bool

+ func NewDB(path string) *DB

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

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

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

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

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

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

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

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

+ func (db *DB) EnsureExists(ctx context.Context) error

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

+ func (db *DB) IsOpen() bool

+ func (db *DB) LTXDir() string

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

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

+ func (db *DB) LastSuccessfulSyncAt() time.Time

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

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

+ func (db *DB) MetaPath() string

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

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

+ func (db *DB) PageSize() int

+ func (db *DB) Path() string

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

+ func (db *DB) RecordSuccessfulSync()

+ func (db *DB) ResetLocalState(ctx context.Context) error

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

+ func (db *DB) SetLogger(logger *slog.Logger)

+ func (db *DB) SetMetaPath(path string)

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

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

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

+ func (db *DB) SyncAndWait(ctx context.Context) error

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

+ func (db *DB) WALPath() string

+ type DBNotReadyError struct

+ Reason string

+ func (e *DBNotReadyError) Error() string

+ func (e *DBNotReadyError) Is(target error) bool

+ type DatabaseSummary struct

+ LastSyncAt *time.Time

+ Path string

+ Status string

+ type ErrorResponse struct

+ Details interface{}

+ Error string

+ type HeartbeatClient struct

+ Interval time.Duration

+ Timeout time.Duration

+ URL string

+ func NewHeartbeatClient(url string, interval time.Duration) *HeartbeatClient

+ func (c *HeartbeatClient) LastPingAt() time.Time

+ func (c *HeartbeatClient) Ping(ctx context.Context) error

+ func (c *HeartbeatClient) RecordPing()

+ func (c *HeartbeatClient) ShouldPing() bool

+ type InfoResponse struct

+ DatabaseCount int

+ PID int

+ StartedAt time.Time

+ UptimeSeconds int64

+ Version string

+ type IntegrityCheckMode int

+ const IntegrityCheckFull

+ const IntegrityCheckNone

+ const IntegrityCheckQuick

+ type LTXError struct

+ Err error

+ Hint string

+ Level int

+ MaxTXID uint64

+ MinTXID uint64

+ Op string

+ Path string

+ func NewLTXError(op, path string, level int, minTXID, maxTXID uint64, err error) *LTXError

+ func (e *LTXError) Error() string

+ func (e *LTXError) IsAutoRecoverable() bool

+ func (e *LTXError) Unwrap() error

+ type Lease struct

+ ETag string

+ ExpiresAt time.Time

+ Generation int64

+ Owner string

+ func (l *Lease) IsExpired() bool

+ func (l *Lease) TTL() time.Duration

+ type LeaseExistsError struct

+ ExpiresAt time.Time

+ Owner string

+ func (e *LeaseExistsError) Error() string

+ type Leaser interface

+ AcquireLease func(ctx context.Context) (*Lease, error)

+ ReleaseLease func(ctx context.Context, lease *Lease) error

+ RenewLease func(ctx context.Context, lease *Lease) (*Lease, error)

+ Type func() string

+ type ListResponse struct

+ Databases []DatabaseSummary

+ type PosV3 struct

+ Generation string

+ Index int

+ Offset int64

+ func (p PosV3) IsZero() bool

+ func (p PosV3) String() string

+ type PrevFrameMismatchError struct

+ Err error

+ func (e *PrevFrameMismatchError) Error() string

+ func (e *PrevFrameMismatchError) Unwrap() error

+ type RegisterDatabaseRequest struct

+ Path string

+ ReplicaURL string

+ type RegisterDatabaseResponse struct

+ Path string

+ Status string

+ type Replica struct

+ AgeIdentities []age.Identity

+ AgeRecipients []age.Recipient

+ AutoRecoverEnabled bool

+ Client ReplicaClient

+ MonitorEnabled bool

+ SyncInterval time.Duration

+ func NewReplica(db *DB) *Replica

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

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

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

+ func (r *Replica) DB() *DB

+ func (r *Replica) DecryptionEnabled() bool

+ func (r *Replica) EncryptionEnabled() bool

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

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

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

+ func (r *Replica) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)

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

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

+ func (r *Replica) RestoreV3(ctx context.Context, opt RestoreOptions) error

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

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

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

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

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

+ func (r *Replica) TimeBoundsV3(ctx context.Context, client ReplicaClientV3) (createdAt, updatedAt time.Time, err error)

+ func (r *Replica) ValidateLevel(ctx context.Context, level int) ([]ValidationError, error)

+ func (r *Replica) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, rd io.Reader) (*ltx.FileInfo, error)

+ type ReplicaClient interface

+ DeleteAll func(ctx context.Context) error

+ DeleteLTXFiles func(ctx context.Context, a []*ltx.FileInfo) error

+ Init func(ctx context.Context) error

+ LTXFiles func(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)

+ OpenLTXFile func(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)

+ SetLogger func(logger *slog.Logger)

+ Type func() string

+ WriteLTXFile func(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error)

+ func NewReplicaClientFromURL(rawURL string) (ReplicaClient, error)

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

+ type ReplicaClientV3 interface

+ GenerationsV3 func(ctx context.Context) ([]string, error)

+ OpenSnapshotV3 func(ctx context.Context, generation string, index int) (io.ReadCloser, error)

+ OpenWALSegmentV3 func(ctx context.Context, generation string, index int, offset int64) (io.ReadCloser, error)

+ SnapshotsV3 func(ctx context.Context, generation string) ([]SnapshotInfoV3, error)

+ WALSegmentsV3 func(ctx context.Context, generation string) ([]WALSegmentInfoV3, error)

+ type RestoreOptions struct

+ Follow bool

+ FollowInterval time.Duration

+ IntegrityCheck IntegrityCheckMode

+ OutputPath string

+ Parallelism int

+ TXID ltx.TXID

+ Timestamp time.Time

+ func NewRestoreOptions() RestoreOptions

+ type Server struct

+ PathExpander func(string) (string, error)

+ SocketPath string

+ SocketPerms uint32

+ Version string

+ func NewServer(store *Store) *Server

+ func (s *Server) Close() error

+ func (s *Server) Start() error

+ type SnapshotInfoV3 struct

+ CreatedAt time.Time

+ Generation string

+ Index int

+ Size int64

+ func (info SnapshotInfoV3) Pos() PosV3

+ type SocketConfig struct

+ Enabled bool

+ Path string

+ Permissions uint32

+ func DefaultSocketConfig() SocketConfig

+ type StartRequest struct

+ Path string

+ Timeout int

+ type StartResponse struct

+ Path string

+ Status string

+ type StopRequest struct

+ Path string

+ Timeout int

+ type StopResponse struct

+ Path string

+ Status string

+ type Store struct

+ CompactionMonitorEnabled bool

+ Heartbeat *HeartbeatClient

+ HeartbeatCheckInterval time.Duration

+ L0Retention time.Duration

+ L0RetentionCheckInterval time.Duration

+ Logger *slog.Logger

+ RetentionEnabled bool

+ ShutdownSyncInterval time.Duration

+ ShutdownSyncTimeout time.Duration

+ SnapshotInterval time.Duration

+ SnapshotRetention time.Duration

+ ValidationInterval time.Duration

+ VerifyCompaction bool

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

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

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

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

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

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

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

+ func (s *Store) FindDB(path string) *DB

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

+ func (s *Store) RegisterDB(db *DB) error

+ func (s *Store) SetDone(done <-chan struct{})

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

+ func (s *Store) SetRetentionEnabled(v bool)

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

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

+ func (s *Store) SetVerifyCompaction(v bool)

+ func (s *Store) SnapshotLevel() *CompactionLevel

+ func (s *Store) SyncDB(ctx context.Context, path string, wait bool) (SyncDBResult, error)

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

+ func (s *Store) Validate(ctx context.Context) (*ValidationResult, error)

+ type SyncDBResult struct

+ Changed bool

+ ReplicatedTXID uint64

+ TXID uint64

+ type SyncRequest struct

+ Path string

+ Timeout int

+ Wait bool

+ type SyncResponse struct

+ Path string

+ ReplicatedTXID uint64

+ Status string

+ TXID uint64

+ type SyncStatus struct

+ InSync bool

+ LocalTXID ltx.TXID

+ RemoteTXID ltx.TXID

+ type TXIDResponse struct

+ TXID uint64

+ type UnregisterDatabaseRequest struct

+ Path string

+ Timeout int

+ type UnregisterDatabaseResponse struct

+ Path string

+ Status string

+ type ValidationError struct

+ CurrFile *ltx.FileInfo

+ Level int

+ Message string

+ PrevFile *ltx.FileInfo

+ Type string

+ type ValidationResult struct

+ Errors []ValidationError

+ Valid bool

+ type WALReader struct

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

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

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

+ func (r *WALReader) Offset() int64

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

+ func (r *WALReader) PageSize() uint32

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

+ type WALSegmentInfoV3 struct

+ CreatedAt time.Time

+ Generation string

+ Index int

+ Offset int64

+ Size int64

+ func (info WALSegmentInfoV3) Pos() PosV3