statesync

Documentation

Index

• Constants
• Variables
• func ChainRPCClient(server string) (*rpchttp.HTTP, error)
• func RestoreDB(ctx context.Context, snapshot io.Reader, ...) error
• func ToABCIApplySnapshotChunkResponse(err error) abciTypes.ResponseApplySnapshotChunk_Result
• func ToABCIOfferSnapshotResponse(err error) abciTypes.ResponseOfferSnapshot_Result
• type DBConfig
• type DBSnapshotter
• type Snapshot
  • func GetLatestSnapshotInfo(ctx context.Context, client cometClient.ABCIClient) (*Snapshot, error)
  • func (s *Snapshot) MarshalBinary() ([]byte, error)
  • func (s *Snapshot) SaveAs(file string) error
  • func (s *Snapshot) UnmarshalBinary(data []byte) error
• type SnapshotConfig
• type SnapshotStore
  • func NewSnapshotStore(cfg *SnapshotConfig, dbCfg *DBConfig, logger log.Logger) (*SnapshotStore, error)
  • func (s *SnapshotStore) CreateSnapshot(ctx context.Context, height uint64, snapshotID string, ...) error
  • func (s *SnapshotStore) IsSnapshotDue(height uint64) bool
  • func (s *SnapshotStore) ListSnapshots() []*Snapshot
  • func (s *SnapshotStore) LoadSnapshotChunk(height uint64, format uint32, chunkIdx uint32) ([]byte, error)
  • func (s *SnapshotStore) RegisterSnapshot(snapshot *Snapshot) error
• type Snapshotter
  • func NewSnapshotter(cfg *DBConfig, dir string, logger log.Logger) *Snapshotter
  • func (s *Snapshotter) CreateSnapshot(ctx context.Context, height uint64, snapshotID string, ...) (*Snapshot, error)
• type StateSyncer
  • func NewStateSyncer(ctx context.Context, cfg *DBConfig, snapshotDir string, providers []string, ...) (*StateSyncer, error)
  • func (ss *StateSyncer) ApplySnapshotChunk(ctx context.Context, chunk []byte, index uint32) (bool, error)
  • func (ss *StateSyncer) OfferSnapshot(ctx context.Context, snapshot *Snapshot) error
• type Streamer
  • func NewStreamer(numChunks uint32, chunkDir string, logger log.Logger) *Streamer
  • func (s *Streamer) Close() error
  • func (s *Streamer) Next() error
  • func (s *Streamer) Read(p []byte) (n int, err error)

Constants

const (
	DefaultSnapshotFormat = 0

	CreateSchema   = "CREATE SCHEMA"
	CreateTable    = "CREATE TABLE"
	CreateFunction = "CREATE FUNCTION"
)

const (
	ABCISnapshotQueryPath        = "/snapshot/height"
	ABCILatestSnapshotHeightPath = "/snapshot/latest"
)

const HashLen int = 32

Variables

var (
	ErrStateSyncInProgress    = fmt.Errorf("statesync already in progress")
	ErrStateSyncNotInProgress = fmt.Errorf("statesync not in progress")

	ErrAbortSnapshot             = fmt.Errorf("abort snapshot")
	ErrRejectSnapshot            = fmt.Errorf("reject snapshot")
	ErrUnsupportedSnapshotFormat = fmt.Errorf("unsupported snapshot format")
	ErrInvalidSnapshot           = fmt.Errorf("invalid snapshot")

	ErrAbortSnapshotChunk   = fmt.Errorf("abort snapshot chunk")
	ErrRetrySnapshotChunk   = fmt.Errorf("retry snapshot chunk")   // retries without refetching the chunk
	ErrRefetchSnapshotChunk = fmt.Errorf("refetch snapshot chunk") // retries after refetching the chunk
	ErrRejectSnapshotChunk  = fmt.Errorf("reject snapshot chunk")
)

Functions

func ChainRPCClient

func ChainRPCClient(server string) (*rpchttp.HTTP, error)

rpcClient sets up a new RPC client

func RestoreDB

func RestoreDB(ctx context.Context, snapshot io.Reader,
	dbName, dbUser, dbPass, dbHost, dbPort string,
	snapshotHash []byte, logger log.Logger) error

RestoreDB restores the database from the logical sql dump using psql command It also validates the snapshot hash, before restoring the database

func ToABCIApplySnapshotChunkResponse

func ToABCIApplySnapshotChunkResponse(err error) abciTypes.ResponseApplySnapshotChunk_Result

func ToABCIOfferSnapshotResponse

func ToABCIOfferSnapshotResponse(err error) abciTypes.ResponseOfferSnapshot_Result

Types

type DBConfig

type DBConfig struct {
	DBUser string
	DBPass string
	DBHost string
	DBPort string
	DBName string
}

type DBSnapshotter

type DBSnapshotter interface {
	CreateSnapshot(ctx context.Context, height uint64, snapshotID string, schemas, excludeTables []string, excludeTableData []string) (*Snapshot, error)
}

type Snapshot

type Snapshot struct {
	Height       uint64          `json:"height"`
	Format       uint32          `json:"format"`
	ChunkHashes  [][HashLen]byte `json:"chunk_hashes"`
	ChunkCount   uint32          `json:"chunk_count"`
	SnapshotHash []byte          `json:"hash"`
	SnapshotSize uint64          `json:"size"`
}

Snapshot is the header of a snapshot file representing the snapshot of the database at a certain height. It contains the height, format, chunk count, hash, size, and name of the snapshot. WARNING: This struct CAN NOT be changed without breaking functionality, since it is used for communication between nodes.

func GetLatestSnapshotInfo

func GetLatestSnapshotInfo(ctx context.Context, client cometClient.ABCIClient) (*Snapshot, error)

GetLatestSnapshotHeight queries the trusted snapshot providers to get the latest snapshot height.

func (*Snapshot) MarshalBinary

func (s *Snapshot) MarshalBinary() ([]byte, error)

func (*Snapshot) SaveAs

func (s *Snapshot) SaveAs(file string) error

SaveAs saves the snapshot header to a file.

func (*Snapshot) UnmarshalBinary

func (s *Snapshot) UnmarshalBinary(data []byte) error

type SnapshotConfig

type SnapshotConfig struct {
	// Snapshot store configuration
	SnapshotDir     string
	MaxSnapshots    int
	RecurringHeight uint64
}

type SnapshotStore

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

func NewSnapshotStore

func NewSnapshotStore(cfg *SnapshotConfig, dbCfg *DBConfig, logger log.Logger) (*SnapshotStore, error)

func (*SnapshotStore) CreateSnapshot

func (s *SnapshotStore) CreateSnapshot(ctx context.Context, height uint64, snapshotID string, schemas, excludedTables []string, excludeTableData []string) error

CreateSnapshot creates a new snapshot at the given height and snapshot ID. SnapshotStore ensures that the number of snapshots does not exceed the maximum configured snapshots. If exceeds, it deletes the oldest snapshot. It takes a list of schemas, excludedTables and excludeTableData args to specify the contents of the snapshot. schemas: list of schemas to include in the snapshot excludedTables: list of tables to exclude from the snapshot excludeTableData: list of tables to include schema but exclude data from the snapshot

func (*SnapshotStore) IsSnapshotDue

func (s *SnapshotStore) IsSnapshotDue(height uint64) bool

IsSnapshotDue checks if a snapshot is due at the given height.

func (*SnapshotStore) ListSnapshots

func (s *SnapshotStore) ListSnapshots() []*Snapshot

List snapshots lists all the registered snapshots in the snapshot store.

func (*SnapshotStore) LoadSnapshotChunk

func (s *SnapshotStore) LoadSnapshotChunk(height uint64, format uint32, chunkIdx uint32) ([]byte, error)

LoadSnapshotChunk loads a snapshot chunk at the given height and chunk index of given format. It returns the snapshot chunk as a byte slice of max size 16MB. errors if the chunk of chunkIndex corresponding to snapshot at given height and format does not exist.

func (*SnapshotStore) RegisterSnapshot

func (s *SnapshotStore) RegisterSnapshot(snapshot *Snapshot) error

RegisterSnapshot registers the existing snapshot in the snapshot store. It ensures that the number of snapshots does not exceed the maximum configured snapshots. If exceeds, it deletes the oldest snapshot.

type Snapshotter

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

func NewSnapshotter

func NewSnapshotter(cfg *DBConfig, dir string, logger log.Logger) *Snapshotter

func (*Snapshotter) CreateSnapshot

func (s *Snapshotter) CreateSnapshot(ctx context.Context, height uint64, snapshotID string, schemas, excludeTables []string, excludeTableData []string) (*Snapshot, error)

CreateSnapshot creates a snapshot at the given height and snapshotID

type StateSyncer

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

func NewStateSyncer

func NewStateSyncer(ctx context.Context, cfg *DBConfig, snapshotDir string, providers []string, db sql.ReadTxMaker, logger log.Logger) (*StateSyncer, error)

NewStateSyncer will initialize the state syncer that enables the node to receive and validate snapshots from the network and initialize the database state. It takes the database configuration, snapshot directory, and the trusted snapshot providers. Trusted snapshot providers are special nodes in the network trusted by the nodes and have snapshot creation enabled. These nodes are responsible for creating and validating snapshots.

func (*StateSyncer) ApplySnapshotChunk

func (ss *StateSyncer) ApplySnapshotChunk(ctx context.Context, chunk []byte, index uint32) (bool, error)

ApplySnapshotChunk accepts a chunk and stores it on disk for later processing if valid If all chunks are received, it starts the process of restoring the database

func (*StateSyncer) OfferSnapshot

func (ss *StateSyncer) OfferSnapshot(ctx context.Context, snapshot *Snapshot) error

OfferSnapshot checks if the snapshot is valid and kicks off the state sync process accepted snapshot is stored on disk for later processing

type Streamer

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

Utility to stream chunks of a snapshot

func NewStreamer

func NewStreamer(numChunks uint32, chunkDir string, logger log.Logger) *Streamer

func (*Streamer) Close

func (s *Streamer) Close() error

func (*Streamer) Next

func (s *Streamer) Next() error

Next opens the next chunk file for streaming

func (*Streamer) Read

func (s *Streamer) Read(p []byte) (n int, err error)

Read reads from the current chunk file If the current chunk is exhausted, it opens the next chunk file until all chunks are read