persistence

Documentation

Index

• Constants
• Variables
• type ConnectionPoolOptions
• type DBKey
  • func NewDBKey(senderTimestamp uint64, receiverTimestamp uint64, pubsubTopic string, ...) *DBKey
  • func (k *DBKey) Bytes() []byte
• type DBOption
  • func DefaultOptions() []DBOption
  • func WithDB(db *sql.DB) DBOption
  • func WithDriver(driverName string, datasourceName string, ...) DBOption
  • func WithMigrations(migrationFn MigrationFn) DBOption
  • func WithRetentionPolicy(maxMessages int, maxDuration time.Duration) DBOption
• type DBStore
  • func NewDBStore(reg prometheus.Registerer, log *zap.Logger, options ...DBOption) (*DBStore, error)
  • func (d *DBStore) Count() (int, error)
  • func (d *DBStore) GetAll() ([]StoredMessage, error)
  • func (d *DBStore) GetStoredMessage(row *sql.Rows) (StoredMessage, error)
  • func (d *DBStore) MostRecentTimestamp() (int64, error)
  • func (d *DBStore) Put(env *protocol.Envelope) error
  • func (d *DBStore) Query(query *pb.HistoryQuery) (*pb.Index, []StoredMessage, error)
  • func (d *DBStore) Start(ctx context.Context, timesource timesource.Timesource) error
  • func (d *DBStore) Stop()
  • func (d *DBStore) Validate(env *protocol.Envelope) error
• type Hash
• type MessageProvider
• type Metrics
• type MigrationFn
• type Queries
  • func CreateQueries(tbl string, db *sql.DB) *Queries
  • func (q Queries) Delete() string
  • func (q Queries) Exists() string
  • func (q Queries) Get() string
  • func (q Queries) GetSize() string
  • func (q Queries) Limit() string
  • func (q Queries) Offset() string
  • func (q Queries) Prefix() string
  • func (q Queries) Put() string
  • func (q Queries) Query() string
• type StoredMessage

Constants

const (
	TimestampLength   = 8
	HashLength        = 32
	DigestLength      = HashLength
	PubsubTopicLength = HashLength
	DBKeyLength       = TimestampLength + PubsubTopicLength + DigestLength
)

const MaxTimeVariance = time.Duration(20) * time.Second

MaxTimeVariance is the maximum duration in the future allowed for a message timestamp

const WALMode = "wal"

WALMode for sqlite.

Variables

var ErrFutureMessage = errors.New("message timestamp in the future")

ErrFutureMessage indicates that a message with timestamp in future was requested to be stored

var (
	// ErrInvalidByteSize is returned when DBKey can't be created
	// from a byte slice because it has invalid length.
	ErrInvalidByteSize = errors.New("byte slice has invalid length")
)

var ErrInvalidCursor = errors.New("invalid cursor")

ErrInvalidCursor indicates that an invalid cursor has been passed to access store

var ErrMessageTooOld = errors.New("message too old")

ErrMessageTooOld indicates that a message that was too old was requested to be stored.

Types

type ConnectionPoolOptions

type ConnectionPoolOptions struct {
	MaxOpenConnections    int
	MaxIdleConnections    int
	ConnectionMaxLifetime time.Duration
	ConnectionMaxIdleTime time.Duration
}

ConnectionPoolOptions is the options to be used for DB connection pooling

type DBKey

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

DBKey key to be stored in a db.

func NewDBKey

func NewDBKey(senderTimestamp uint64, receiverTimestamp uint64, pubsubTopic string, digest []byte) *DBKey

NewDBKey creates a new DBKey with the given values.

func (*DBKey) Bytes

func (k *DBKey) Bytes() []byte

Bytes returns a bytes representation of the DBKey.

type DBOption

type DBOption func(*DBStore) error

DBOption is an optional setting that can be used to configure the DBStore

func DefaultOptions

func DefaultOptions() []DBOption

DefaultOptions returns the default DBoptions to be used.

func WithDB

func WithDB(db *sql.DB) DBOption

WithDB is a DBOption that lets you use any custom *sql.DB with a DBStore.

func WithDriver

func WithDriver(driverName string, datasourceName string, connectionPoolOptions ...ConnectionPoolOptions) DBOption

WithDriver is a DBOption that will open a *sql.DB connection

func WithMigrations

func WithMigrations(migrationFn MigrationFn) DBOption

WithMigrations is a DBOption used to determine if migrations should be executed, and what driver to use

func WithRetentionPolicy

func WithRetentionPolicy(maxMessages int, maxDuration time.Duration) DBOption

WithRetentionPolicy is a DBOption that specifies the max number of messages to be stored and duration before they're removed from the message store

type DBStore

type DBStore struct {
	MessageProvider
	// contains filtered or unexported fields
}

DBStore is a MessageProvider that has a *sql.DB connection

func NewDBStore

func NewDBStore(reg prometheus.Registerer, log *zap.Logger, options ...DBOption) (*DBStore, error)

Creates a new DB store using the db specified via options. It will create a messages table if it does not exist and clean up records according to the retention policy used

func (*DBStore) Count

func (d *DBStore) Count() (int, error)

Count returns the number of rows in the message table

func (*DBStore) GetAll

func (d *DBStore) GetAll() ([]StoredMessage, error)

GetAll returns all the stored WakuMessages

func (*DBStore) GetStoredMessage

func (d *DBStore) GetStoredMessage(row *sql.Rows) (StoredMessage, error)

GetStoredMessage is a helper function used to convert a `*sql.Rows` into a `StoredMessage`

func (*DBStore) MostRecentTimestamp

func (d *DBStore) MostRecentTimestamp() (int64, error)

MostRecentTimestamp returns an unix timestamp with the most recent senderTimestamp in the message table

func (*DBStore) Put

func (d *DBStore) Put(env *protocol.Envelope) error

Put inserts a WakuMessage into the DB

func (*DBStore) Query

func (d *DBStore) Query(query *pb.HistoryQuery) (*pb.Index, []StoredMessage, error)

Query retrieves messages from the DB

func (*DBStore) Start

func (d *DBStore) Start(ctx context.Context, timesource timesource.Timesource) error

Start starts the store server functionality

func (*DBStore) Stop

func (d *DBStore) Stop()

Stop closes a DB connection

func (*DBStore) Validate

func (d *DBStore) Validate(env *protocol.Envelope) error

Validate validates the message to be stored against possible fradulent conditions.

type Hash

type Hash [HashLength]byte

type MessageProvider

type MessageProvider interface {
	GetAll() ([]StoredMessage, error)
	Validate(env *protocol.Envelope) error
	Put(env *protocol.Envelope) error
	Query(query *pb.HistoryQuery) ([]StoredMessage, error)
	MostRecentTimestamp() (int64, error)
	Start(ctx context.Context, timesource timesource.Timesource) error
	Stop()
}

MessageProvider is an interface that provides access to store/retrieve messages from a persistence store.

type Metrics

type Metrics interface {
	RecordMessage(num int)
	RecordError(err metricsErrCategory)
	RecordInsertDuration(duration time.Duration)
	RecordQueryDuration(duration time.Duration)
}

Metrics exposes the functions required to update prometheus metrics for archive protocol

type MigrationFn

type MigrationFn func(db *sql.DB) error

type Queries

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

Queries are the SQL queries for a given table.

func CreateQueries

func CreateQueries(tbl string, db *sql.DB) *Queries

CreateQueries Function creates a set of queries for an SQL table. Note: Do not use this function to create queries for a table, rather use <rdb>.NewQueries to create table as well as queries.

func (Queries) Delete

func (q Queries) Delete() string

Delete returns the query for deleting a row.

func (Queries) Exists

func (q Queries) Exists() string

Exists returns the query for determining if a row exists.

func (Queries) Get

func (q Queries) Get() string

Get returns the query for getting a row.

func (Queries) GetSize

func (q Queries) GetSize() string

GetSize returns the query for determining the size of a value.

func (Queries) Limit

func (q Queries) Limit() string

Limit returns the query fragment for limiting results.

func (Queries) Offset

func (q Queries) Offset() string

Offset returns the query fragment for returning rows from a given offset.

func (Queries) Prefix

func (q Queries) Prefix() string

Prefix returns the query fragment for getting a rows with a key prefix.

func (Queries) Put

func (q Queries) Put() string

Put returns the query for putting a row.

func (Queries) Query

func (q Queries) Query() string

Query returns the query for getting multiple rows.

type StoredMessage

type StoredMessage struct {
	ID           []byte
	PubsubTopic  string
	ReceiverTime int64
	Message      *wpb.WakuMessage
}

StoredMessage is the format of the message stored in persistence store