chore(jobsdb): add support for schema migration changesets to run always

jobsdb/jobsdb.go

@@ -839,10 +839,50 @@ func (jd *HandleT) init() {
 	// so that we can be sure that all the necessary tables are created and considered to be in
 	// the latest schema version, before rudder-migrator starts introducing new tables.
 	jd.dsListLock.WithLock(func(l lock.LockToken) {
-		switch jd.ownerType {
-		case Write, ReadWrite:
-			jd.setupDatabaseTables(l, jd.clearAll)
+
+		// TODO: we need to make sure that no other process will be able to add or remove tables
+		// during the migration, otherwise schema migration may panic and db will be left in a dirty
+		// state.
+		// An advisory lock, shared between schema migration, internal migration and
+		// add new ds loop might be necessary.
+
+		writer := jd.ownerType == Write || jd.ownerType == ReadWrite
+		if writer && jd.clearAll {
+			jd.dropDatabaseTables(l)
+		}
+		templateData := func() map[string]interface{} {
+			// Important: if jobsdb type is acting as a writer then refreshDSList
+			// doesn't return the full list of datasets, only the rightmost two.
+			// But we need to run the schema migration against all datasets, no matter
+			// whether jobsdb is a writer or not.
+			datasets := getDSList(jd, jd.dbHandle, jd.tablePrefix)
+
+			datasetIndices := make([]string, 0)
+			for _, dataset := range datasets {
+				datasetIndices = append(datasetIndices, dataset.Index)
+			}
+
+			return map[string]interface{}{
+				"Prefix":   jd.tablePrefix,
+				"Datasets": datasetIndices,
+			}
+		}()
+
+		if writer {
+			jd.setupDatabaseTables(templateData)
 		}
+
+		// Run changesets that should always run for both writer and reader jobsdbs.
+		//
+		// When running separate gw and processor instances we cannot control the order of execution
+		// and we cannot guarantee that while gw migration is running or after it is complete, processor
+		// will not create new tables using the old schema.
+		//
+		// Changesets that run always can help in such cases, by bringing non-migrated tables into a usable state.
+		jd.runAlwaysChangesets(templateData)
+
+		// finally refresh the dataset list to make sure [datasetList] field is populated
+		jd.refreshDSList(l)
 	})
 }
 

jobsdb/setup.go

@@ -1,13 +1,11 @@
 package jobsdb
 
 import (
-	"database/sql"
 	"fmt"
 
 	"github.com/rudderlabs/rudder-server/config"
 	"github.com/rudderlabs/rudder-server/jobsdb/internal/lock"
 	migrator "github.com/rudderlabs/rudder-server/services/sql-migrator"
-	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
 // SchemaMigrationTable returns the table name used for storing current schema version.
@@ -20,50 +18,30 @@ func (jd *HandleT) SchemaMigrationTable() string {
 // The following data are passed to JobsDB migration templates:
 // - Prefix: The table prefix used by this jobsdb instance.
 // - Datasets: Array of existing dataset indices.
-// If clearAll is set to true, all existing jobsdb tables will be removed first.
-func (jd *HandleT) setupDatabaseTables(l lock.LockToken, clearAll bool) {
-	if clearAll {
-		jd.dropDatabaseTables(l)
-	}
-
-	// Important: if jobsdb type is acting as a writer then refreshDSList
-	// doesn't return the full list of datasets, only the rightmost two.
-	// But we need to run the schema migration against all datasets, no matter
-	// whether jobsdb is a writer or not.
-	datasets := getDSList(jd, jd.dbHandle, jd.tablePrefix)
-
-	datasetIndices := make([]string, 0)
-	for _, dataset := range datasets {
-		datasetIndices = append(datasetIndices, dataset.Index)
-	}
-
-	templateData := map[string]interface{}{
-		"Prefix":   jd.tablePrefix,
-		"Datasets": datasetIndices,
-	}
-
-	psqlInfo := misc.GetConnectionString()
-	db, err := sql.Open("postgres", psqlInfo)
-	if err != nil {
-		panic(fmt.Errorf("error DB for migrate open: %w", err))
-	}
-
-	defer func() { _ = db.Close() }()
-
+func (jd *HandleT) setupDatabaseTables(templateData map[string]interface{}) {
 	// setup migrator with appropriate schema migrations table
 	m := &migrator.Migrator{
-		Handle:                     db,
+		Handle:                     jd.dbHandle,
 		MigrationsTable:            jd.SchemaMigrationTable(),
 		ShouldForceSetLowerVersion: config.GetBool("SQLMigrator.forceSetLowerVersion", true),
 	}
-
 	// execute any necessary migrations
-	err = m.MigrateFromTemplates("jobsdb", templateData)
-	if err != nil {
+	if err := m.MigrateFromTemplates("jobsdb", templateData); err != nil {
 		panic(fmt.Errorf("error while migrating '%v' jobsdb tables: %w", jd.tablePrefix, err))
 	}
-	// finally refresh the dataset list to make sure [datasetList] field is populated
-	jd.refreshDSList(l)
+}
+
+func (jd *HandleT) runAlwaysChangesets(templateData map[string]interface{}) {
+	// setup migrator with appropriate schema migrations table
+	m := &migrator.Migrator{
+		Handle:          jd.dbHandle,
+		MigrationsTable: fmt.Sprintf("%s_runalways_migrations", jd.tablePrefix),
+		RunAlways:       true,
+	}
+	// execute any necessary migrations
+	if err := m.MigrateFromTemplates("jobsdb_always", templateData); err != nil {
+		panic(fmt.Errorf("error while running changesets that run always in '%s' jobsdb tables: %w", jd.tablePrefix, err))
+	}
 }
 
 func (jd *HandleT) dropDatabaseTables(l lock.LockToken) {

services/sql-migrator/migrator.go

@@ -31,6 +31,9 @@ type Migrator struct {
 	// Indicates if migration version should be force reset to latest on file in case of revert to lower version
 	// Eg. DB has v3 set in MigrationsTable but latest version in MigrationsDir is v2
 	ShouldForceSetLowerVersion bool
+
+	// Indicates if all migrations should be run ignoring the current version in MigrationsTable
+	RunAlways bool
 }
 
 var pkgLogger logger.Logger
@@ -146,12 +149,17 @@ func (m *Migrator) MigrateFromTemplates(templatesDir string, context interface{}
 	}
 
 	if m.ShouldForceSetLowerVersion {
-		err := m.forceSetLowerVersion(migration, sourceDriver, destinationDriver)
-		if err != nil {
+		if err := m.forceSetLowerVersion(migration, sourceDriver, destinationDriver); err != nil {
 			return err
 		}
 	}
 
+	if m.RunAlways {
+		if err := migration.Force(-1); err != nil {
+			return fmt.Errorf("force migration version to 0: %w", err)
+		}
+	}
+
 	err = migration.Up()
 	if err != nil && err != migrate.ErrNoChange { // migrate library reports that no change was required, using ErrNoChange
 		return fmt.Errorf("run migration from template directory %q, %w", templatesDir, err)

sql/migrations/jobsdb_always/000010_add_index_job_status_table.up.tmpl

@@ -0,0 +1,4 @@
+{{range .Datasets}}
+    CREATE INDEX IF NOT EXISTS "idx_{{$.Prefix}}_job_status_{{.}}_jid_id" ON "{{$.Prefix}}_job_status_{{.}}" (job_id asc,id desc);
+    CREATE OR REPLACE VIEW "v_last_{{$.Prefix}}_job_status_{{.}}" AS SELECT DISTINCT ON (job_id) * FROM "{{$.Prefix}}_job_status_{{.}}" ORDER BY job_id ASC,id DESC;
+{{end}}