syncer

Documentation

Index

• Constants
• Variables
• func OpenEnv(l logrus.FieldLogger, lc config.LMDB) (env *lmdb.Env, err error)
• type ErrEntry
  • func (e ErrEntry) Error() string
  • func (e ErrEntry) Unwrap() error
• type InstanceSet
  • func NewInstanceSet() *InstanceSet
  • func (s *InstanceSet) Add(name string)
  • func (s *InstanceSet) CleanDisappeared(seen []string) (cleaned []string)
  • func (s *InstanceSet) Contains(name string) bool
  • func (s *InstanceSet) Done() bool
  • func (s *InstanceSet) List() []string
  • func (s *InstanceSet) Remove(name string)
  • func (s *InstanceSet) String() string
• type NativeIterator
  • func NewNativeIterator(formatVersion uint32, compatVersion uint32, dbiMsg *snapshot.DBI, ...) (*NativeIterator, error)
  • func (it *NativeIterator) Clean(oldval []byte) (val []byte, err error)
  • func (it *NativeIterator) Merge(oldval []byte) (val []byte, err error)
  • func (it *NativeIterator) Next() (key []byte, err error)
• type Options
• type PlainIterator
  • func (it *PlainIterator) Clean(oldval []byte) (val []byte, err error)
  • func (it *PlainIterator) Merge(oldval []byte) (val []byte, err error)
  • func (it *PlainIterator) Next() (key []byte, err error)
• type Syncer
  • func New(name string, env *lmdb.Env, st simpleblob.Interface, c config.Config, ...) (*Syncer, error)
  • func (s *Syncer) LoadOnce(ctx context.Context, env *lmdb.Env, instance string, update snapshot.Update, ...) (txnID header.TxnID, localChanged bool, err error)
  • func (s *Syncer) SendOnce(ctx context.Context, env *lmdb.Env) (txnID header.TxnID, err error)
  • func (s *Syncer) Sync(ctx context.Context) error

Constants

const (
	LMDBMaxKeySize        = 511
	DupSortHackMaxKeySize = 255
)

const (
	// SyncDBIPrefix is the shared DBI name prefix for all special tables that
	// must not be synced.
	SyncDBIPrefix = "_sync"
	// SyncDBIShadowPrefix is the DBI name prefix of shadow databases.
	SyncDBIShadowPrefix = "_sync_shadow_"
)

const (
	// AllowedShadowDBIFlagsMask is the set of LMDB DBI flags that we transfer
	// to shadow DBIs.
	// MDB_INTEGERKEY needs to be transferred for proper ordering of shadow DBIs.
	AllowedShadowDBIFlagsMask = dbiflags.IntegerKey
)

const (
	// MaxConsecutiveSnapshotLoads are the maximum number of snapshot to
	// load before we break for snapshotting local changes, if local changes
	// exist.
	MaxConsecutiveSnapshotLoads = 10
)

Variables

var (
	ErrNoTxnID = errors.New("no TxnID set on iterator")
)

Functions

func OpenEnv

func OpenEnv(l logrus.FieldLogger, lc config.LMDB) (env *lmdb.Env, err error)

OpenEnv opens the LMDB env with the right options

Types

type ErrEntry

type ErrEntry struct {
	DBIName string
	Key     []byte
	Err     error
}

ErrEntry is returned when an entry is invalid, for example due to a missing or invalid header.

func (ErrEntry) Error

func (e ErrEntry) Error() string

func (ErrEntry) Unwrap

func (e ErrEntry) Unwrap() error

type InstanceSet

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

InstanceSet is a set of instances that we are still waiting for

func NewInstanceSet

func NewInstanceSet() *InstanceSet

func (*InstanceSet) Add

func (s *InstanceSet) Add(name string)

func (*InstanceSet) CleanDisappeared

func (s *InstanceSet) CleanDisappeared(seen []string) (cleaned []string)

func (*InstanceSet) Contains

func (s *InstanceSet) Contains(name string) bool

func (*InstanceSet) Done

func (s *InstanceSet) Done() bool

func (*InstanceSet) List

func (s *InstanceSet) List() []string

func (*InstanceSet) Remove

func (s *InstanceSet) Remove(name string)

func (*InstanceSet) String

func (s *InstanceSet) String() string

type NativeIterator

type NativeIterator struct {
	DBIMsg               *snapshot.DBI    // DBI contents as raw values without header
	DefaultTimestampNano header.Timestamp // Timestamp to add to entries that do not have one
	TxnID                header.TxnID     // Current write TxnID (required)
	FormatVersion        uint32           // Snapshot FormatVersion
	HeaderPaddingBlock   bool             // Extra padding block for testing
	DeletedCutoff        header.Timestamp // Older deleted entries are considered stale
	// contains filtered or unexported fields
}

NativeIterator iterates over a snapshot DBI and updates the LMDB with values that are prefixed with a native header. This iterator has two uses: * Merge the main database into a shadow database with a default timestamp. * Merge a remote snapshot with the timestamp values into a DBI with headers. The LMDB values the iterator operates on MUST always have a header. If no header is present, an error is returned.

func NewNativeIterator

func NewNativeIterator(
	formatVersion uint32,
	compatVersion uint32,
	dbiMsg *snapshot.DBI,
	defaultTS header.Timestamp,
	txnID header.TxnID,
	deletedCutoff header.Timestamp,
) (*NativeIterator, error)

func (*NativeIterator) Clean

func (it *NativeIterator) Clean(oldval []byte) (val []byte, err error)

func (*NativeIterator) Merge

func (it *NativeIterator) Merge(oldval []byte) (val []byte, err error)

Merge compares the old LMDB value currently stored and the current iterator value from the dump, and decides which value the LMDB should take. The LMDB entries are always prefixed with a header.

func (*NativeIterator) Next

func (it *NativeIterator) Next() (key []byte, err error)

type Options

type Options struct {
	// ReceiveOnly prevents writing snapshots, we will only receive them
	ReceiveOnly bool
	// Events are used to publish events to
	Events *events.Events
	// Hooks can be used to update data at certain points.
	Hooks *hooks.Hooks
}

type PlainIterator

type PlainIterator struct {
	DBIMsg *snapshot.DBI // LMDB contents (timestamp is ignored)
	// contains filtered or unexported fields
}

PlainIterator iterates over a snapshot of a shadow database for insertion into the main database without the timestamp header.

func (*PlainIterator) Clean

func (it *PlainIterator) Clean(oldval []byte) (val []byte, err error)

func (*PlainIterator) Merge

func (it *PlainIterator) Merge(oldval []byte) (val []byte, err error)

func (*PlainIterator) Next

func (it *PlainIterator) Next() (key []byte, err error)

type Syncer

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

func New

func New(name string, env *lmdb.Env, st simpleblob.Interface, c config.Config, lc config.LMDB, opt Options) (*Syncer, error)

func (*Syncer) LoadOnce

func (s *Syncer) LoadOnce(ctx context.Context, env *lmdb.Env, instance string, update snapshot.Update, lastTxnID header.TxnID) (txnID header.TxnID, localChanged bool, err error)

func (*Syncer) SendOnce

func (s *Syncer) SendOnce(ctx context.Context, env *lmdb.Env) (txnID header.TxnID, err error)

func (*Syncer) Sync

func (s *Syncer) Sync(ctx context.Context) error

Sync opens the env and starts the two-way sync loop.