Documentation
¶
Index ¶
- Constants
- Variables
- func TrimName(name string) string
- func WriteStreamFrame(w io.Writer, f StreamFrame) error
- type Client
- type DB
- func (db *DB) ApplyLTX(ctx context.Context, path string) error
- func (db *DB) CommitJournal(mode JournalMode) error
- func (db *DB) CreateJournal() (*os.File, error)
- func (db *DB) DatabasePath() string
- func (db *DB) EnforceRetention(ctx context.Context, minTime time.Time) error
- func (db *DB) ID() uint32
- func (db *DB) InWriteTx() bool
- func (db *DB) JournalPath() string
- func (db *DB) LTXDir() string
- func (db *DB) LTXPath(minTXID, maxTXID uint64) string
- func (db *DB) Name() string
- func (db *DB) Open() error
- func (db *DB) OpenLTXFile(txID uint64) (*os.File, error)
- func (db *DB) PageSize() uint32
- func (db *DB) Path() string
- func (db *DB) PendingLock() *RWMutex
- func (db *DB) Pos() Pos
- func (db *DB) ReadLTXDir() ([]fs.DirEntry, error)
- func (db *DB) ReservedLock() *RWMutex
- func (db *DB) SharedLock() *RWMutex
- func (db *DB) TXID() uint64
- func (db *DB) WriteDatabase(f *os.File, data []byte, offset int64) error
- func (db *DB) WriteJournal(f *os.File, data []byte, offset int64) error
- func (db *DB) WriteSnapshotTo(ctx context.Context, dst io.Writer) (header ltx.Header, trailer ltx.Trailer, err error)
- type DBStreamFrame
- type FileType
- type Invalidator
- type JournalMode
- type LTXStreamFrame
- type Lease
- type Leaser
- type LockType
- type Pos
- type PrimaryInfo
- type RWMutex
- func (rw *RWMutex) CanLock() bool
- func (rw *RWMutex) CanRLock() bool
- func (rw *RWMutex) Lock(ctx context.Context) (*RWMutexGuard, error)
- func (rw *RWMutex) RLock(ctx context.Context) (*RWMutexGuard, error)
- func (rw *RWMutex) State() RWMutexState
- func (rw *RWMutex) TryLock() *RWMutexGuard
- func (rw *RWMutex) TryRLock() *RWMutexGuard
- type RWMutexGuard
- type RWMutexState
- type StaticLease
- type StaticLeaser
- type Store
- func (s *Store) Candidate() bool
- func (s *Store) Close() error
- func (s *Store) CreateDB(name string) (*DB, *os.File, error)
- func (s *Store) DB(id uint32) *DB
- func (s *Store) DBByName(name string) *DB
- func (s *Store) DBDir() string
- func (s *Store) DBPath(id uint32) string
- func (s *Store) DBs() []*DB
- func (s *Store) EnforceRetention(ctx context.Context) (err error)
- func (s *Store) ForceCreateDB(id uint32, name string) (*DB, error)
- func (s *Store) ID() string
- func (s *Store) IsPrimary() bool
- func (s *Store) MarkDirty(dbID uint32)
- func (s *Store) Open() error
- func (s *Store) Path() string
- func (s *Store) PosMap() map[uint32]Pos
- func (s *Store) PrimaryCtx(ctx context.Context) context.Context
- func (s *Store) PrimaryInfo() *PrimaryInfo
- func (s *Store) ReadyCh() chan struct{}
- func (s *Store) Subscribe() *Subscriber
- func (s *Store) Unsubscribe(sub *Subscriber)
- type StoreVar
- type StreamFrame
- type StreamFrameType
- type Subscriber
Constants ¶
const ( SQLITE_DATABASE_HEADER_STRING = "SQLite format 3\x00" /// Magic header string that identifies a SQLite journal header. /// https://www.sqlite.org/fileformat.html#the_rollback_journal SQLITE_JOURNAL_HEADER_STRING = "\xd9\xd5\x05\xf9\x20\xa1\x63\xd7" // Location of the database size, in pages, in the main database file. SQLITE_DATABASE_SIZE_OFFSET = 28 )
const ( LockTypePending = 0x40000000 LockTypeReserved = 0x40000001 )
const ( WALHeaderSize = 32 WALFrameHeaderSize = 24 WALIndexHeaderSize = 136 )
SQLite constants
const ( PENDING_BYTE = 0x40000000 RESERVED_BYTE = (PENDING_BYTE + 1) SHARED_FIRST = (PENDING_BYTE + 2) SHARED_SIZE = 510 )
SQLite rollback journal lock constants.
const ( WAL_WRITE_LOCK = 120 WAL_CKPT_LOCK = 121 WAL_RECOVER_LOCK = 122 WAL_READ_LOCK0 = 123 WAL_READ_LOCK1 = 124 WAL_READ_LOCK2 = 125 WAL_READ_LOCK3 = 126 WAL_READ_LOCK4 = 127 )
SQLite WAL lock constants.
const ( F_OFD_GETLK = 36 F_OFD_SETLK = 37 F_OFD_SETLKW = 38 )
Open file description lock constants.
const ( JournalModeDelete = "DELETE" JournalModeTruncate = "TRUNCATE" JournalModePersist = "PERSIST" JournalModeWAL = "WAL" )
const ( FileTypeNone = FileType(iota) FileTypeDatabase FileTypeJournal FileTypeWAL FileTypeSHM FileTypePos )
Database file types.
const ( StreamFrameTypeDB = StreamFrameType(1) StreamFrameTypeLTX = StreamFrameType(2) )
const ( RWMutexStateUnlocked = iota RWMutexStateExclusive )
const ( DefaultRetentionDuration = 1 * time.Minute DefaultRetentionMonitorInterval = 1 * time.Minute )
Default store settings.
const IDLength = 24
IDLength is the length of a node ID, in bytes.
const RWMutexInterval = 10 * time.Microsecond
RWMutexInterval is the time between reattempting lock acquisition.
Variables ¶
var ( ErrDatabaseNotFound = fmt.Errorf("database not found") ErrDatabaseExists = fmt.Errorf("database already exists") ErrNoPrimary = errors.New("no primary") ErrPrimaryExists = errors.New("primary exists") ErrLeaseExpired = errors.New("lease expired") ErrReadOnlyReplica = fmt.Errorf("read only replica") )
LiteFS errors
Functions ¶
func WriteStreamFrame ¶
func WriteStreamFrame(w io.Writer, f StreamFrame) error
WriteStreamFrame writes the stream type & frame to the writer.
Types ¶
type Client ¶
type Client interface {
// Stream starts a long-running connection to stream changes from another node.
Stream(ctx context.Context, rawurl string, id string, posMap map[uint32]Pos) (io.ReadCloser, error)
}
Client represents a client for connecting to other LiteFS nodes.
type DB ¶
type DB struct {
// Returns the current time. Used for mocking time in tests.
Now func() time.Time
// contains filtered or unexported fields
}
DB represents a SQLite database.
func (*DB) CommitJournal ¶
func (db *DB) CommitJournal(mode JournalMode) error
CommitJournal deletes the journal file which commits or rolls back the transaction.
func (*DB) CreateJournal ¶
CreateJournal creates a new journal file on disk.
func (*DB) DatabasePath ¶ added in v0.1.1
DatabasePath returns the path to the underlying database file.
func (*DB) EnforceRetention ¶ added in v0.2.0
EnforceRetention removes all LTX files created before minTime.
func (*DB) JournalPath ¶ added in v0.1.1
JournalPath returns the path to the underlying journal file.
func (*DB) OpenLTXFile ¶
OpenLTXFile returns a file handle to an LTX file that contains the given TXID.
func (*DB) PendingLock ¶
func (*DB) ReadLTXDir ¶ added in v0.2.0
ReadLTXDir returns DirEntry for every LTX file.
func (*DB) ReservedLock ¶
func (*DB) SharedLock ¶
func (*DB) WriteDatabase ¶
WriteDatabase writes data to the main database file.
func (*DB) WriteJournal ¶
WriteJournal writes data to the rollback journal file.
type DBStreamFrame ¶
DBStreamFrame represents a frame with basic database information. This is sent at the beginning of the stream and when a new database is created.
func (*DBStreamFrame) Type ¶
func (*DBStreamFrame) Type() StreamFrameType
Type returns the type of stream frame.
type Invalidator ¶ added in v0.1.1
type Invalidator interface {
InvalidateDB(db *DB, offset, size int64) error
InvalidatePos(db *DB) error
}
Invalidator is a callback for the store to use to invalidate the kernel page cache.
type LTXStreamFrame ¶
type LTXStreamFrame struct {
}
func (*LTXStreamFrame) Type ¶
func (*LTXStreamFrame) Type() StreamFrameType
Type returns the type of stream frame.
type Lease ¶
type Lease interface {
RenewedAt() time.Time
TTL() time.Duration
// Renew attempts to reset the TTL on the lease.
// Returns ErrLeaseExpired if the lease has expired or was deleted.
Renew(ctx context.Context) error
// Close attempts to remove the lease from the server.
Close() error
}
Lease represents an acquired lease from a Leaser.
type Leaser ¶
type Leaser interface {
io.Closer
AdvertiseURL() string
// Acquire attempts to acquire the lease to become the primary.
Acquire(ctx context.Context) (Lease, error)
// PrimaryInfo attempts to read the current primary data.
// Returns ErrNoPrimary if no primary currently has the lease.
PrimaryInfo(ctx context.Context) (PrimaryInfo, error)
}
Leaser represents an API for obtaining a lease for leader election.
type LockType ¶
type LockType int
LockType represents a SQLite lock type.
func ParseLockRange ¶
ParseLockRange returns a list of SQLite locks that are within a range.
type PrimaryInfo ¶ added in v0.2.0
type PrimaryInfo struct {
Hostname string `json:"hostname"`
AdvertiseURL string `json:"advertise-url"`
}
PrimaryInfo is the JSON object stored in the Consul lease value.
func (*PrimaryInfo) Clone ¶ added in v0.2.0
func (info *PrimaryInfo) Clone() *PrimaryInfo
Clone returns a copy of info.
type RWMutex ¶
type RWMutex struct {
// contains filtered or unexported fields
}
RWMutex is a reader/writer mutual exclusion lock. It wraps the sync package to provide additional capabilities such as lock upgrades & downgrades. It only supports TryLock() & TryRLock() as that is what's supported by our FUSE file system.
func (*RWMutex) Lock ¶
func (rw *RWMutex) Lock(ctx context.Context) (*RWMutexGuard, error)
Lock attempts to obtain a exclusive lock on rw. Returns an error if ctx is done.
func (*RWMutex) RLock ¶
func (rw *RWMutex) RLock(ctx context.Context) (*RWMutexGuard, error)
RLock attempts to obtain a shared lock on rw. Returns an error if ctx is done.
func (*RWMutex) State ¶
func (rw *RWMutex) State() RWMutexState
State returns whether the mutex has a exclusive lock, one or more shared locks, or if the mutex is unlocked.
func (*RWMutex) TryLock ¶
func (rw *RWMutex) TryLock() *RWMutexGuard
TryLock tries to lock the mutex for writing and returns a guard if it succeeds.
func (*RWMutex) TryRLock ¶
func (rw *RWMutex) TryRLock() *RWMutexGuard
TryRLock tries to lock rw for reading and reports whether it succeeded.
type RWMutexGuard ¶
type RWMutexGuard struct {
// contains filtered or unexported fields
}
RWMutexGuard is a reference to a held lock.
func (*RWMutexGuard) CanLock ¶
func (g *RWMutexGuard) CanLock() bool
CanLock returns true if the guard can become an exclusive lock.
func (*RWMutexGuard) RLock ¶
func (g *RWMutexGuard) RLock()
RLock downgrades the lock from an exclusive lock to a shared lock. This is a no-op if the lock is already a shared lock.
func (*RWMutexGuard) TryLock ¶
func (g *RWMutexGuard) TryLock() bool
TryLock upgrades the lock from a shared lock to an exclusive lock. This is a no-op if the lock is already an exclusive lock.
func (*RWMutexGuard) Unlock ¶
func (g *RWMutexGuard) Unlock()
Unlock unlocks the underlying mutex. Guard must be discarded after Unlock().
type RWMutexState ¶
type RWMutexState int
RWMutexState represents the lock state of an RWMutexGuard.
func (RWMutexState) String ¶ added in v0.2.0
func (s RWMutexState) String() string
String returns the string representation of the state.
type StaticLease ¶ added in v0.2.0
type StaticLease struct {
// contains filtered or unexported fields
}
StaticLease represents a lease for a fixed primary.
func (*StaticLease) Close ¶ added in v0.2.0
func (l *StaticLease) Close() error
func (*StaticLease) Renew ¶ added in v0.2.0
func (l *StaticLease) Renew(ctx context.Context) error
Renew is a no-op.
func (*StaticLease) RenewedAt ¶ added in v0.2.0
func (l *StaticLease) RenewedAt() time.Time
RenewedAt returns the Unix epoch in UTC.
func (*StaticLease) TTL ¶ added in v0.2.0
func (l *StaticLease) TTL() time.Duration
TTL returns the duration until the lease expires which is a time well into the future.
type StaticLeaser ¶ added in v0.2.0
type StaticLeaser struct {
// contains filtered or unexported fields
}
StaticLeaser always returns a lease to a static primary.
func NewStaticLeaser ¶ added in v0.2.0
func NewStaticLeaser(isPrimary bool, hostname, advertiseURL string) *StaticLeaser
NewStaticLeaser returns a new instance of StaticLeaser.
func (*StaticLeaser) Acquire ¶ added in v0.2.0
func (l *StaticLeaser) Acquire(ctx context.Context) (Lease, error)
Acquire returns a lease if this node is the static primary. Otherwise returns ErrPrimaryExists.
func (*StaticLeaser) AdvertiseURL ¶ added in v0.2.0
func (l *StaticLeaser) AdvertiseURL() string
AdvertiseURL returns the primary URL if this is the primary. Otherwise returns blank.
func (*StaticLeaser) Close ¶ added in v0.2.0
func (l *StaticLeaser) Close() (err error)
Close is a no-op.
func (*StaticLeaser) IsPrimary ¶ added in v0.2.0
func (l *StaticLeaser) IsPrimary() bool
IsPrimary returns true if the current node is the primary.
func (*StaticLeaser) PrimaryInfo ¶ added in v0.2.0
func (l *StaticLeaser) PrimaryInfo(ctx context.Context) (PrimaryInfo, error)
PrimaryInfo returns the primary's info. Returns ErrNoPrimary if the node is the primary.
type Store ¶
type Store struct {
// Client used to connect to other LiteFS instances.
Client Client
// Leaser manages the lease that controls leader election.
Leaser Leaser
// Length of time to retain LTX files.
RetentionDuration time.Duration
RetentionMonitorInterval time.Duration
// Callback to notify kernel of file changes.
Invalidator Invalidator
// contains filtered or unexported fields
}
Store represents a collection of databases.
func (*Store) Candidate ¶ added in v0.2.0
Candidate returns true if store is eligible to be the primary.
func (*Store) CreateDB ¶
CreateDB creates a new database with the given name. The returned file handle must be closed by the caller. Returns an error if a database with the same name already exists.
func (*Store) DBByName ¶
DBByName returns a database by name. Returns nil if the database does not exist.
func (*Store) EnforceRetention ¶ added in v0.2.0
EnforceRetention enforces retention of LTX files on all databases.
func (*Store) ForceCreateDB ¶
ForceCreateDB creates a database with the given ID & name. This occurs when replicating from a primary server.
func (*Store) ID ¶ added in v0.2.0
ID returns the unique identifier for this instance. Available after Open(). Persistent across restarts if underlying storage is persistent.
func (*Store) PrimaryCtx ¶ added in v0.2.0
PrimaryCtx wraps ctx with another context that will cancel when no longer primary.
func (*Store) PrimaryInfo ¶ added in v0.2.0
func (s *Store) PrimaryInfo() *PrimaryInfo
PrimaryInfo returns info about the current primary.
func (*Store) ReadyCh ¶ added in v0.2.0
func (s *Store) ReadyCh() chan struct{}
ReadyCh returns a channel that is closed once the store has become primary or once it has connected to the primary.
func (*Store) Subscribe ¶
func (s *Store) Subscribe() *Subscriber
Subscribe creates a new subscriber for store changes.
func (*Store) Unsubscribe ¶
func (s *Store) Unsubscribe(sub *Subscriber)
Unsubscribe removes a subscriber from the store.
type StreamFrame ¶
type StreamFrame interface {
io.ReaderFrom
io.WriterTo
Type() StreamFrameType
}
func ReadStreamFrame ¶
func ReadStreamFrame(r io.Reader) (StreamFrame, error)
ReadStreamFrame reads a the stream type & frame from the reader.
type StreamFrameType ¶
type StreamFrameType uint32
type Subscriber ¶
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber subscribes to changes to databases in the store.
It implements a set of "dirty" databases instead of a channel of all events as clients can be slow and we don't want to cause channels to back up. It is the responsibility of the caller to determine the state changes which is usually just checking the position of the client versus the store's database.
func (*Subscriber) Close ¶
func (s *Subscriber) Close() error
Close removes the subscriber from the store.
func (*Subscriber) DirtySet ¶
func (s *Subscriber) DirtySet() map[uint32]struct{}
DirtySet returns a set of database IDs that have changed since the last call to DirtySet(). This call clears the set.
func (*Subscriber) MarkDirty ¶
func (s *Subscriber) MarkDirty(dbID uint32)
MarkDirty marks a database ID as dirty.
func (*Subscriber) NotifyCh ¶
func (s *Subscriber) NotifyCh() <-chan struct{}
NotifyCh returns a channel that receives a value when the dirty set has changed.