add db migrations

sql/database.go

@@ -115,7 +115,18 @@ func Open(uri string, opts ...Opt) (*Database, error) {
 	if config.enableLatency {
 		db.latency = newQueryLatency()
 	}
+	for i := 0; i < config.connections; i++ {
+		conn := pool.Get(context.Background())
+		if err := registerFunctions(conn); err != nil {
+			return nil, err
+		}
+		pool.Put(conn)
+	}
 	if config.migrations != nil {
+		before, err := version(db)
+		if err != nil {
+			return nil, err
+		}
 		tx, err := db.Tx(context.Background())
 		if err != nil {
 			return nil, err
@@ -128,13 +139,11 @@ func Open(uri string, opts ...Opt) (*Database, error) {
 		if err != nil {
 			return nil, err
 		}
-	}
-	for i := 0; i < config.connections; i++ {
-		conn := pool.Get(context.Background())
-		if err := registerFunctions(conn); err != nil {
-			return nil, err
+		if before == 3 {
+			if err := Vacuum(db); err != nil {
+				return nil, err
+			}
 		}
-		defer pool.Put(conn)
 	}
 	return db, nil
 }

sql/functions.go

@@ -4,6 +4,9 @@ import (
 	"fmt"
 
 	sqlite "github.com/go-llsqlite/crawshaw"
+
+	"github.com/spacemeshos/go-spacemesh/codec"
+	"github.com/spacemeshos/go-spacemesh/common/types"
 )
 
 func registerFunctions(conn *sqlite.Conn) error {
@@ -16,5 +19,21 @@ func registerFunctions(conn *sqlite.Conn) error {
 	}, nil, nil); err != nil {
 		return fmt.Errorf("registering add_uint64: %w", err)
 	}
+	// function to prune active set from old ballots
+	if err := conn.CreateFunction("prune_actives", true, 1, func(ctx sqlite.Context, values ...sqlite.Value) {
+		var ballot types.Ballot
+		if err := codec.Decode(values[0].Blob(), &ballot); err != nil {
+			ctx.ResultError(err)
+		} else {
+			ballot.ActiveSet = nil
+			if blob, err := codec.Encode(&ballot); err != nil {
+				ctx.ResultError(err)
+			} else {
+				ctx.ResultBlob(blob)
+			}
+		}
+	}, nil, nil); err != nil {
+		return fmt.Errorf("registering prune_actives: %w", err)
+	}
 	return nil
 }

sql/migrations.go

@@ -23,6 +23,17 @@ type migration struct {
 // Migrations is interface for migrations provider.
 type Migrations func(Executor) error
 
+func version(db Executor) (int, error) {
+	var current int
+	if _, err := db.Exec("PRAGMA user_version;", nil, func(stmt *Statement) bool {
+		current = stmt.ColumnInt(0)
+		return true
+	}); err != nil {
+		return 0, fmt.Errorf("read user_version %w", err)
+	}
+	return current, nil
+}
+
 func embeddedMigrations(db Executor) error {
 	var migrations []migration
 	fs.WalkDir(embedded, "migrations", func(path string, d fs.DirEntry, err error) error {
@@ -62,13 +73,9 @@ func embeddedMigrations(db Executor) error {
 		return migrations[i].order < migrations[j].order
 	})
 
-	var current int
-
-	if _, err := db.Exec("PRAGMA user_version;", nil, func(stmt *Statement) bool {
-		current = stmt.ColumnInt(0)
-		return true
-	}); err != nil {
-		return fmt.Errorf("read user_version %w", err)
+	current, err := version(db)
+	if err != nil {
+		return err
 	}
 
 	for _, m := range migrations {

sql/migrations/0004_next.sql

@@ -0,0 +1,5 @@
+DELETE FROM proposals WHERE layer < 19000;
+DELETE FROM proposal_transactions WHERE layer < 19000;
+UPDATE certificates SET cert = NULL WHERE layer < 19000;
+UPDATE ballots SET ballot = prune_actives(ballot);
+

sql/migrations_test.go

@@ -15,5 +15,5 @@ func TestMigrationsAppliedOnce(t *testing.T) {
 		return true
 	})
 	require.NoError(t, err)
-	require.Equal(t, version, 3)
+	require.Equal(t, version, 4)
 }

sql/vacuum.go

@@ -0,0 +1,15 @@
+package sql
+
+import (
+	"fmt"
+)
+
+func Vacuum(db Executor) error {
+	if _, err := db.Exec("vacuum", nil, nil); err != nil {
+		return fmt.Errorf("vacuum %w", err)
+	}
+	if _, err := db.Exec("pragma wal_checkpoint(TRUNCATE)", nil, nil); err != nil {
+		return fmt.Errorf("wal checkpoint %w", err)
+	}
+	return nil
+}

sql/vacuum_test.go

@@ -0,0 +1,12 @@
+package sql
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestVacuumDB(t *testing.T) {
+	db := InMemory()
+	require.NoError(t, Vacuum(db))
+}