feat: sqlite vacuum and optional migrations

This commit is contained in:
Richard Ramos 2023-08-08 11:46:32 -04:00 committed by richΛrd
parent b88907c5ee
commit ee2af4646c
12 changed files with 59 additions and 14 deletions

View File

@ -288,6 +288,19 @@ var (
Destination: &options.Store.DatabaseURL,
EnvVars: []string{"WAKUNODE2_STORE_MESSAGE_DB_URL"},
})
StoreMessageDBVacuum = altsrc.NewBoolFlag(&cli.BoolFlag{
Name: "store-message-db-vacuum",
Usage: "Enable database vacuuming at start. Only supported by SQLite database engine.",
Destination: &options.Store.Vacuum,
EnvVars: []string{"WAKUNODE2_STORE_MESSAGE_DB_VACUUM"},
})
StoreMessageDBMigration = altsrc.NewBoolFlag(&cli.BoolFlag{
Name: "store-message-db-migration",
Usage: "Enable database migration at start.",
Destination: &options.Store.Migration,
Value: true,
EnvVars: []string{"WAKUNODE2_STORE_MESSAGE_DB_MIGRATION"},
})
StoreResumePeer = cliutils.NewGenericFlagMultiValue(&cli.GenericFlag{
Name: "store-resume-peer",
Usage: "Peer multiaddress to resume the message store at boot. Option may be repeated",

View File

@ -56,6 +56,8 @@ func main() {
StoreMessageDBURL,
StoreMessageRetentionTime,
StoreMessageRetentionCapacity,
StoreMessageDBVacuum,
StoreMessageDBMigration,
StoreResumePeer,
FilterFlag,
FilterNode,

View File

@ -96,8 +96,11 @@ func Execute(options NodeOptions) {
var db *sql.DB
var migrationFn func(*sql.DB) error
if requiresDB(options) {
db, migrationFn, err = dbutils.ExtractDBAndMigration(options.Store.DatabaseURL)
if requiresDB(options) && options.Store.Migration {
dbSettings := dbutils.DBSettings{
SQLiteVacuum: options.Store.Vacuum,
}
db, migrationFn, err = dbutils.ExtractDBAndMigration(options.Store.DatabaseURL, dbSettings, logger)
failOnErr(err, "Could not connect to DB")
}
@ -211,11 +214,16 @@ func Execute(options NodeOptions) {
var dbStore *persistence.DBStore
if requiresDB(options) {
dbStore, err = persistence.NewDBStore(logger,
dbOptions := []persistence.DBOption{
persistence.WithDB(db),
persistence.WithMigrations(migrationFn), // TODO: refactor migrations out of DBStore, or merge DBStore with rendezvous DB
persistence.WithRetentionPolicy(options.Store.RetentionMaxMessages, options.Store.RetentionTime),
)
}
if options.Store.Migration {
dbOptions = append(dbOptions, persistence.WithMigrations(migrationFn)) // TODO: refactor migrations out of DBStore, or merge DBStore with rendezvous DB
}
dbStore, err = persistence.NewDBStore(logger, dbOptions...)
failOnErr(err, "DBStore")
nodeOpts = append(nodeOpts, node.WithMessageProvider(dbStore))
}

View File

@ -81,6 +81,8 @@ type StoreOptions struct {
RetentionMaxMessages int
ResumeNodes []multiaddr.Multiaddr
Nodes []multiaddr.Multiaddr
Vacuum bool
Migration bool
}
// DNSDiscoveryOptions are settings used for enabling DNS-based discovery

View File

@ -12,7 +12,7 @@ import (
func MemoryDB(t *testing.T) *persistence.DBStore {
var db *sql.DB
db, migration, err := sqlite.NewDB(":memory:")
db, migration, err := sqlite.NewDB(":memory:", false, utils.Logger())
require.NoError(t, err)
dbStore, err := persistence.NewDBStore(utils.Logger(), persistence.WithDB(db), persistence.WithMigrations(migration))

View File

@ -131,7 +131,7 @@ func NewNode(configJSON string) error {
if *config.EnableStore {
var db *sql.DB
var migrationFn func(*sql.DB) error
db, migrationFn, err = dbutils.ExtractDBAndMigration(*config.DatabaseURL)
db, migrationFn, err = dbutils.ExtractDBAndMigration(*config.DatabaseURL, dbutils.DBSettings{SQLiteVacuum: true}, utils.Logger())
if err != nil {
return err
}

View File

@ -11,6 +11,7 @@ import (
"github.com/waku-org/go-waku/waku/persistence"
"github.com/waku-org/go-waku/waku/persistence/migrate"
"github.com/waku-org/go-waku/waku/persistence/sqlite/migrations"
"go.uber.org/zap"
)
func addSqliteURLDefaults(dburl string) string {
@ -56,7 +57,7 @@ func WithDB(dburl string, migrate bool) persistence.DBOption {
}
// NewDB creates a sqlite3 DB in the specified path
func NewDB(dburl string) (*sql.DB, func(*sql.DB) error, error) {
func NewDB(dburl string, shouldVacuum bool, logger *zap.Logger) (*sql.DB, func(*sql.DB) error, error) {
db, err := sql.Open("sqlite3", addSqliteURLDefaults(dburl))
if err != nil {
return nil, nil, err
@ -65,6 +66,17 @@ func NewDB(dburl string) (*sql.DB, func(*sql.DB) error, error) {
// Disable concurrent access as not supported by the driver
db.SetMaxOpenConns(1)
logger.Info("starting sqlite database vacuuming")
if shouldVacuum {
_, err := db.Exec("VACUUM")
if err != nil {
return nil, nil, err
}
}
logger.Info("finished sqlite database vacuuming")
return db, Migrate, nil
}

View File

@ -8,6 +8,7 @@ import (
"github.com/waku-org/go-waku/waku/persistence/postgres"
"github.com/waku-org/go-waku/waku/persistence/sqlite"
"go.uber.org/zap"
)
func validateDBUrl(val string) error {
@ -18,12 +19,19 @@ func validateDBUrl(val string) error {
return nil
}
// DBSettings hold db specific configuration settings required during the db initialization
type DBSettings struct {
SQLiteVacuum bool
}
// ExtractDBAndMigration will return a database connection, and migration function that should be used depending on a database connection string
func ExtractDBAndMigration(databaseURL string) (*sql.DB, func(*sql.DB) error, error) {
func ExtractDBAndMigration(databaseURL string, dbSettings DBSettings, logger *zap.Logger) (*sql.DB, func(*sql.DB) error, error) {
var db *sql.DB
var migrationFn func(*sql.DB) error
var err error
logger = logger.Named("db-setup")
dbURL := ""
if databaseURL != "" {
err := validateDBUrl(databaseURL)
@ -41,7 +49,7 @@ func ExtractDBAndMigration(databaseURL string) (*sql.DB, func(*sql.DB) error, er
dbParams := dbURLParts[1]
switch dbEngine {
case "sqlite3":
db, migrationFn, err = sqlite.NewDB(dbParams)
db, migrationFn, err = sqlite.NewDB(dbParams, dbSettings.SQLiteVacuum, logger)
case "postgresql":
db, migrationFn, err = postgres.NewDB(dbURL)
default:

View File

@ -69,7 +69,7 @@ func TestConnectionStatusChanges(t *testing.T) {
err = node2.Start(ctx)
require.NoError(t, err)
db, migration, err := sqlite.NewDB(":memory:")
db, migration, err := sqlite.NewDB(":memory:", false, utils.Logger())
require.NoError(t, err)
dbStore, err := persistence.NewDBStore(utils.Logger(), persistence.WithDB(db), persistence.WithMigrations(migration))
require.NoError(t, err)

View File

@ -230,7 +230,7 @@ func TestDecoupledStoreFromRelay(t *testing.T) {
subs.Unsubscribe()
// NODE2: Filter Client/Store
db, migration, err := sqlite.NewDB(":memory:")
db, migration, err := sqlite.NewDB(":memory:", false, utils.Logger())
require.NoError(t, err)
dbStore, err := persistence.NewDBStore(utils.Logger(), persistence.WithDB(db), persistence.WithMigrations(migration))
require.NoError(t, err)

View File

@ -12,7 +12,7 @@ import (
func MemoryDB(t *testing.T) *persistence.DBStore {
var db *sql.DB
db, migration, err := sqlite.NewDB(":memory:")
db, migration, err := sqlite.NewDB(":memory:", false, utils.Logger())
require.NoError(t, err)
dbStore, err := persistence.NewDBStore(utils.Logger(), persistence.WithDB(db), persistence.WithMigrations(migration))

View File

@ -47,7 +47,7 @@ func TestRendezvous(t *testing.T) {
require.NoError(t, err)
var db *sql.DB
db, migration, err := sqlite.NewDB(":memory:")
db, migration, err := sqlite.NewDB(":memory:", false, utils.Logger())
require.NoError(t, err)
err = migration(db)