using fix from vocdoni-node indexer

cmd/census3/main.go

@@ -28,19 +28,19 @@ func main() {
 	flag.Parse()
 	log.Init(*logLevel, "stdout", nil)
 
-	database, q, err := db.Init(*dataDir)
+	database, err := db.Init(*dataDir)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	// Start the holder scanner
-	hc, err := service.NewHoldersScanner(database, q, *url)
+	hc, err := service.NewHoldersScanner(database, *url)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	// Start the API
-	err = api.Init(database, q, api.Census3APIConf{
+	err = api.Init(database, api.Census3APIConf{
 		Hostname: "0.0.0.0",
 		Port:     *port,
 		DataDir:  *dataDir,

db/db.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"time"
 
 	_ "github.com/mattn/go-sqlite3"
 	"github.com/pressly/goose/v3"
@@ -15,31 +16,73 @@ import (
 //go:embed migrations/*.sql
 var migrationsFS embed.FS
 
-func Init(dataDir string) (*sql.DB, *queries.Queries, error) {
+// DB struct abstact a safe connection with the database using sqlc queries,
+// sqlite as a database engine and go-sqlite3 as a driver.
+type DB struct {
+	RW *sql.DB
+	RO *sql.DB
+
+	QueriesRW *queries.Queries
+	QueriesRO *queries.Queries
+}
+
+// Close function stops all internal connections to the database
+func (db *DB) Close() error {
+	if err := db.RW.Close(); err != nil {
+		return err
+	}
+	return db.RO.Close()
+}
+
+// Init function starts a datbase using the data path provided as argument. It
+// opens two different connections, one for read only, and another for read and
+// write, with different configurations, optimized for each use case.
+func Init(dataDir string) (*DB, error) {
 	dbFile := filepath.Join(dataDir, "census3.sql")
 	if _, err := os.Stat(dbFile); os.IsNotExist(err) {
 		if err := os.MkdirAll(dataDir, os.ModePerm); err != nil {
-			return nil, nil, fmt.Errorf("error creating a new database file: %w", err)
+			return nil, fmt.Errorf("error creating a new database file: %w", err)
 		}
 	}
-	// open database file
-	fileURI := fmt.Sprintf("file:%s?cache=shared", dbFile)
-	database, err := sql.Open("sqlite3", fileURI)
+	// sqlite doesn't support multiple concurrent writers.
+	// For that reason, rwDB is limited to one open connection.
+	// Per https://github.com/mattn/go-sqlite3/issues/1022#issuecomment-1067353980,
+	// we use WAL to allow multiple concurrent readers at the same time.
+	rwDB, err := sql.Open("sqlite3", fmt.Sprintf("file:%s?mode=rwc&_journal_mode=wal&_txlock=immediate&_synchronous=normal", dbFile))
 	if err != nil {
-		return nil, nil, fmt.Errorf("error opening database: %w", err)
+		return nil, fmt.Errorf("error opening database: %w", err)
 	}
-	// trying to fix "database is locked" issue according to the official
-	// mattn/go-sqlite3 docs: https://github.com/mattn/go-sqlite3/#faq
-	database.SetMaxOpenConns(1)
+	rwDB.SetMaxOpenConns(1)
+	rwDB.SetMaxIdleConns(2)
+	rwDB.SetConnMaxIdleTime(10 * time.Minute)
+	rwDB.SetConnMaxLifetime(time.Hour)
+
+	roDB, err := sql.Open("sqlite3", fmt.Sprintf("file:%s?mode=ro&_journal_mode=wal", dbFile))
+	if err != nil {
+		return nil, fmt.Errorf("error opening database: %w", err)
+	}
+	// Increasing these numbers can allow for more queries to run concurrently,
+	// but it also increases the memory used by sqlite and our connection pool.
+	// Most read-only queries we run are quick enough, so a small number seems OK.
+	roDB.SetMaxOpenConns(10)
+	roDB.SetMaxIdleConns(20)
+	roDB.SetConnMaxIdleTime(5 * time.Minute)
+	roDB.SetConnMaxLifetime(time.Hour)
+
 	// get census3 goose migrations and setup for sqlite3
 	if err := goose.SetDialect("sqlite3"); err != nil {
-		return nil, nil, fmt.Errorf("error setting up driver for sqlite: %w", err)
+		return nil, fmt.Errorf("error setting up driver for sqlite: %w", err)
 	}
 	goose.SetBaseFS(migrationsFS)
 	// perform goose up
-	if err := goose.Up(database, "migrations"); err != nil {
-		return nil, nil, fmt.Errorf("error during goose up: %w", err)
+	if err := goose.Up(rwDB, "migrations"); err != nil {
+		return nil, fmt.Errorf("error during goose up: %w", err)
 	}
 	// init sqlc
-	return database, queries.New(database), nil
+	return &DB{
+		RW:        rwDB,
+		RO:        roDB,
+		QueriesRW: queries.New(rwDB),
+		QueriesRO: queries.New(roDB),
+	}, nil
 }

service/helper_test.go

@@ -43,23 +43,23 @@ var (
 )
 
 type TestDB struct {
-	dir     string
-	db      *sql.DB
-	queries *queries.Queries
+	dir string
+	db  *db.DB
 }
 
 func StartTestDB(t *testing.T) *TestDB {
 	c := qt.New(t)
 
 	dir := t.TempDir()
-	db, q, err := db.Init(dir)
+	db, err := db.Init(dir)
 	c.Assert(err, qt.IsNil)
-	return &TestDB{dir, db, q}
+	return &TestDB{dir, db}
 }
 
 func (testdb *TestDB) Close(t *testing.T) {
 	c := qt.New(t)
-	c.Assert(testdb.db.Close(), qt.IsNil)
+	c.Assert(testdb.db.RW.Close(), qt.IsNil)
+	c.Assert(testdb.db.RO.Close(), qt.IsNil)
 	c.Assert(os.RemoveAll(testdb.dir), qt.IsNil)
 }
 

service/holder_scanner_test.go

@@ -22,24 +22,24 @@ func TestNewHolderScanner(t *testing.T) {
 	testdb := StartTestDB(t)
 	defer testdb.Close(t)
 
-	hs, err := NewHoldersScanner(testdb.db, testdb.queries, web3uri)
+	hs, err := NewHoldersScanner(testdb.db, web3uri)
 	c.Assert(err, qt.IsNil)
 	c.Assert(hs.lastBlock, qt.Equals, uint64(0))
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	_, err = testdb.queries.CreateBlock(ctx, queries.CreateBlockParams{
+	_, err = testdb.db.QueriesRW.CreateBlock(ctx, queries.CreateBlockParams{
 		ID:        1000,
 		Timestamp: "test",
 		RootHash:  []byte("test"),
 	})
 	c.Assert(err, qt.IsNil)
 
-	hs, err = NewHoldersScanner(testdb.db, testdb.queries, web3uri)
+	hs, err = NewHoldersScanner(testdb.db, web3uri)
 	c.Assert(err, qt.IsNil)
 	c.Assert(hs.lastBlock, qt.Equals, uint64(1000))
 
-	_, err = NewHoldersScanner(nil, nil, web3uri)
+	_, err = NewHoldersScanner(nil, web3uri)
 	c.Assert(err, qt.IsNotNil)
 }
 
@@ -52,7 +52,7 @@ func TestHolderScannerStart(t *testing.T) {
 	defer testdb.Close(t)
 
 	twg.Add(1)
-	hs, err := NewHoldersScanner(testdb.db, testdb.queries, web3uri)
+	hs, err := NewHoldersScanner(testdb.db, web3uri)
 	c.Assert(err, qt.IsNil)
 	go func() {
 		hs.Start(ctx)
@@ -69,7 +69,7 @@ func Test_tokenAddresses(t *testing.T) {
 	testdb := StartTestDB(t)
 	defer testdb.Close(t)
 
-	hs, err := NewHoldersScanner(testdb.db, testdb.queries, web3uri)
+	hs, err := NewHoldersScanner(testdb.db, web3uri)
 	c.Assert(err, qt.IsNil)
 
 	res, err := hs.tokenAddresses()
@@ -78,7 +78,7 @@ func Test_tokenAddresses(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	_, err = testdb.queries.CreateToken(ctx, testTokenParams("0x1", "test0",
+	_, err = testdb.db.QueriesRW.CreateToken(ctx, testTokenParams("0x1", "test0",
 		"test0", MonkeysDecimals, 0, MonkeysTotalSupply.Uint64(),
 		uint64(state.CONTRACT_TYPE_ERC20), false))
 	c.Assert(err, qt.IsNil)
@@ -87,7 +87,7 @@ func Test_tokenAddresses(t *testing.T) {
 	c.Assert(err, qt.IsNil)
 	c.Assert(res[common.HexToAddress("0x1")], qt.IsFalse)
 
-	_, err = testdb.queries.CreateToken(ctx, testTokenParams("0x2", "test2",
+	_, err = testdb.db.QueriesRW.CreateToken(ctx, testTokenParams("0x2", "test2",
 		"test3", MonkeysDecimals, 10, MonkeysTotalSupply.Uint64(),
 		uint64(state.CONTRACT_TYPE_ERC20), false))
 	c.Assert(err, qt.IsNil)
@@ -103,13 +103,13 @@ func Test_saveHolders(t *testing.T) {
 	testdb := StartTestDB(t)
 	defer testdb.Close(t)
 
-	hs, err := NewHoldersScanner(testdb.db, testdb.queries, "http://google.com")
+	hs, err := NewHoldersScanner(testdb.db, "http://google.com")
 	c.Assert(err, qt.IsNil)
 
 	th := new(state.TokenHolders).Init(MonkeysAddress, state.CONTRACT_TYPE_ERC20, MonkeysCreationBlock)
 	// no registered token
 	c.Assert(hs.saveHolders(th), qt.ErrorIs, ErrTokenNotExists)
-	_, err = testdb.queries.CreateToken(context.Background(), testTokenParams(
+	_, err = testdb.db.QueriesRW.CreateToken(context.Background(), testTokenParams(
 		MonkeysAddress.String(), MonkeysName, MonkeysSymbol, MonkeysDecimals,
 		MonkeysCreationBlock, MonkeysTotalSupply.Uint64(),
 		uint64(state.CONTRACT_TYPE_ERC20), false))
@@ -124,13 +124,13 @@ func Test_saveHolders(t *testing.T) {
 	// check wrong web3
 	c.Assert(hs.saveHolders(th), qt.IsNotNil)
 	// check new block created
-	_, err = testdb.queries.BlockByID(context.Background(), int64(MonkeysCreationBlock))
+	_, err = testdb.db.QueriesRW.BlockByID(context.Background(), int64(MonkeysCreationBlock))
 	c.Assert(err, qt.ErrorIs, sql.ErrNoRows)
 	// check good web3
 	hs.web3 = web3uri
 	c.Assert(hs.saveHolders(th), qt.IsNil)
 	// check new holders
-	res, err := testdb.queries.TokenHolderByTokenIDAndHolderID(context.Background(),
+	res, err := testdb.db.QueriesRW.TokenHolderByTokenIDAndHolderID(context.Background(),
 		queries.TokenHolderByTokenIDAndHolderIDParams{
 			TokenID:  MonkeysAddress.Bytes(),
 			HolderID: holderAddr.Bytes(),
@@ -140,7 +140,7 @@ func Test_saveHolders(t *testing.T) {
 	// check update holders
 	th.Append(holderAddr, holderBalance)
 	c.Assert(hs.saveHolders(th), qt.IsNil)
-	res, err = testdb.queries.TokenHolderByTokenIDAndHolderID(context.Background(),
+	res, err = testdb.db.QueriesRW.TokenHolderByTokenIDAndHolderID(context.Background(),
 		queries.TokenHolderByTokenIDAndHolderIDParams{
 			TokenID:  MonkeysAddress.Bytes(),
 			HolderID: holderAddr.Bytes(),
@@ -151,7 +151,7 @@ func Test_saveHolders(t *testing.T) {
 	// check delete holders
 	th.Append(holderAddr, big.NewInt(-24))
 	c.Assert(hs.saveHolders(th), qt.IsNil)
-	_, err = testdb.queries.TokenHolderByTokenIDAndHolderID(context.Background(),
+	_, err = testdb.db.QueriesRW.TokenHolderByTokenIDAndHolderID(context.Background(),
 		queries.TokenHolderByTokenIDAndHolderIDParams{
 			TokenID:  MonkeysAddress.Bytes(),
 			HolderID: holderAddr.Bytes(),
@@ -165,7 +165,7 @@ func Test_scanHolders(t *testing.T) {
 	testdb := StartTestDB(t)
 	defer testdb.Close(t)
 
-	hs, err := NewHoldersScanner(testdb.db, testdb.queries, web3uri)
+	hs, err := NewHoldersScanner(testdb.db, web3uri)
 	c.Assert(err, qt.IsNil)
 
 	// token does not exists
@@ -174,7 +174,7 @@ func Test_scanHolders(t *testing.T) {
 	_, err = hs.scanHolders(ctx1, MonkeysAddress)
 	c.Assert(err, qt.IsNotNil)
 
-	_, err = testdb.queries.CreateToken(context.Background(), testTokenParams(
+	_, err = testdb.db.QueriesRW.CreateToken(context.Background(), testTokenParams(
 		MonkeysAddress.String(), MonkeysName, MonkeysSymbol, MonkeysDecimals, MonkeysCreationBlock, 10,
 		uint64(state.CONTRACT_TYPE_ERC20), false))
 	c.Assert(err, qt.IsNil)
@@ -184,7 +184,7 @@ func Test_scanHolders(t *testing.T) {
 	_, err = hs.scanHolders(ctx2, MonkeysAddress)
 	c.Assert(err, qt.IsNil)
 
-	res, err := testdb.queries.TokenHoldersByTokenID(context.Background(), MonkeysAddress.Bytes())
+	res, err := testdb.db.QueriesRW.TokenHoldersByTokenID(context.Background(), MonkeysAddress.Bytes())
 	c.Assert(err, qt.IsNil)
 	for _, holder := range res {
 		balance, ok := MonkeysHolders[common.BytesToAddress(holder.ID)]
@@ -199,18 +199,18 @@ func Test_calcTokenCreationBlock(t *testing.T) {
 	testdb := StartTestDB(t)
 	defer testdb.Close(t)
 
-	hs, err := NewHoldersScanner(testdb.db, testdb.queries, web3uri)
+	hs, err := NewHoldersScanner(testdb.db, web3uri)
 	c.Assert(err, qt.IsNil)
 	c.Assert(hs.calcTokenCreationBlock(context.Background(), MonkeysAddress), qt.IsNotNil)
 
-	_, err = testdb.queries.CreateToken(context.Background(), testTokenParams(
+	_, err = testdb.db.QueriesRW.CreateToken(context.Background(), testTokenParams(
 		MonkeysAddress.String(), MonkeysName, MonkeysSymbol, MonkeysDecimals,
 		MonkeysCreationBlock, MonkeysTotalSupply.Uint64(),
 		uint64(state.CONTRACT_TYPE_ERC20), false))
 	c.Assert(err, qt.IsNil)
 
 	c.Assert(hs.calcTokenCreationBlock(context.Background(), MonkeysAddress), qt.IsNil)
-	token, err := testdb.queries.TokenByID(context.Background(), MonkeysAddress.Bytes())
+	token, err := testdb.db.QueriesRW.TokenByID(context.Background(), MonkeysAddress.Bytes())
 	c.Assert(err, qt.IsNil)
 	c.Assert(uint64(token.CreationBlock.Int32), qt.Equals, MonkeysCreationBlock)
 }