Merge pull request #2422 from hashicorp/hugoamvieira-controller-dbswap

If the database url is updated on Boundary's config and then a SIGHUP
triggered, we now connect to the new database, ensure it's OK to use
and use that one going forwards instead.

This allows a controller to be deployed in a more dynamic infrastructure
where, for example, database credentials could change, without having
to restart the controller.

This operation is done atomically to prevent errors and inconsistency. 
A timeout is present so that if the object is swapped after the 
existing one has been looked up it isn't immediately invalidated. 
The timeout is a maximum; a ticker once a second that looks at
stats to determine how many database connections are still in use 
and then closes the driver as soon as none are in use.

internal/cmd/base/dev.go

@@ -93,7 +93,7 @@ func (b *Server) CreateDevDatabase(ctx context.Context, opt ...Option) error {
 		b.Info["dev database container"] = strings.TrimPrefix(container, "/")
 	}
 
-	if err := b.ConnectToDatabase(ctx, dialect); err != nil {
+	if err := b.OpenAndSetServerDatabase(ctx, dialect); err != nil {
 		if c != nil {
 			err = multierror.Append(err, c())
 		}

internal/cmd/base/servers.go

@@ -678,11 +678,26 @@ func (b *Server) RunShutdownFuncs() error {
 	return mErr.ErrorOrNil()
 }
 
-func (b *Server) ConnectToDatabase(ctx context.Context, dialect string) error {
+// OpenAndSetServerDatabase calls OpenDatabase and sets its result *db.DB to the Server's
+// `Database` field.
+func (b *Server) OpenAndSetServerDatabase(ctx context.Context, dialect string) error {
+	dbase, err := b.OpenDatabase(ctx, dialect, b.DatabaseUrl)
+	if err != nil {
+		return err
+	}
+	b.Database = dbase
+	return nil
+}
+
+// OpenDatabase creates a database connection with the given URL and returns it to the caller.
+// It supports various configuration options - The values must be set on the Server object
+// beforehand.
+func (b *Server) OpenDatabase(ctx context.Context, dialect, url string) (*db.DB, error) {
 	dbType, err := db.StringToDbType(dialect)
 	if err != nil {
-		return fmt.Errorf("unable to create db object with dialect %s: %w", dialect, err)
+		return nil, fmt.Errorf("unable to create db object with dialect %s: %w", dialect, err)
 	}
+
 	opts := []db.Option{
 		db.WithMaxOpenConnections(b.DatabaseMaxOpenConnections),
 		db.WithMaxIdleConnections(b.DatabaseMaxIdleConnections),
@@ -691,12 +706,13 @@ func (b *Server) ConnectToDatabase(ctx context.Context, dialect string) error {
 	if os.Getenv("BOUNDARY_DISABLE_GORM_FORMATTER") == "" {
 		opts = append(opts, db.WithGormFormatter(b.Logger))
 	}
-	dbase, err := db.Open(ctx, dbType, b.DatabaseUrl, opts...)
+
+	dbase, err := db.Open(ctx, dbType, url, opts...)
 	if err != nil {
-		return fmt.Errorf("unable to create db object with dialect %s: %w", dialect, err)
+		return nil, fmt.Errorf("unable to create db object with dialect %s: %w", dialect, err)
 	}
-	b.Database = dbase
-	return nil
+
+	return dbase, nil
 }
 
 func (b *Server) CreateGlobalKmsKeys(ctx context.Context) error {

internal/cmd/commands/database/funcs_test.go

@@ -226,7 +226,7 @@ func TestVerifyOplogIsEmpty(t *testing.T) {
 	cmd := InitCommand{Server: base.NewServer(base.NewCommand(cli.NewMockUi()))}
 
 	cmd.DatabaseUrl = u
-	require.NoError(t, cmd.ConnectToDatabase(ctx, dialect))
+	require.NoError(t, cmd.OpenAndSetServerDatabase(ctx, dialect))
 
 	assert.NoError(t, cmd.verifyOplogIsEmpty(ctx))
 }

internal/cmd/commands/database/init.go

@@ -287,7 +287,7 @@ func (c *InitCommand) Run(args []string) (retCode int) {
 		return base.CommandUserError
 	}
 	// Everything after is done with normal database URL and is affecting actual data
-	if err := c.ConnectToDatabase(c.Context, dialect); err != nil {
+	if err := c.OpenAndSetServerDatabase(c.Context, dialect); err != nil {
 		c.UI.Error(fmt.Errorf("Error connecting to database after migrations: %w", err).Error())
 		return base.CommandCliError
 	}

internal/cmd/commands/server/controller_db_swap_test.go

@@ -0,0 +1,459 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/cmd/base"
+	"github.com/hashicorp/boundary/internal/cmd/config"
+	"github.com/hashicorp/boundary/internal/daemon/controller"
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/testing/dbtest"
+	"github.com/mitchellh/cli"
+	"github.com/stretchr/testify/require"
+)
+
+const dbSwapConfig = `
+disable_mlock = true
+
+telemetry {
+	prometheus_retention_time = "24h"
+	disable_hostname = true
+}
+
+controller {
+	name = "test-controller"
+	description = "A default controller created for tests"
+	database {
+		url = "%s"
+	}
+}
+
+kms "aead" {
+	purpose = "root"
+	aead_type = "aes-gcm"
+	key = "%s"
+	key_id = "global_root"
+}
+
+kms "aead" {
+	purpose = "worker-auth"
+	aead_type = "aes-gcm"
+	key = "%s"
+	key_id = "global_worker-auth"
+}
+
+kms "aead" {
+	purpose = "recovery"
+	aead_type = "aes-gcm"
+	key = "%s"
+	key_id = "global_recovery"
+}
+
+listener "tcp" {
+	purpose = "api"
+	address = "127.0.0.1:9500"
+	tls_disable = true
+}
+
+listener "tcp" {
+	address = "127.0.0.1:9501"
+	purpose = "cluster"
+}
+`
+
+const getDatabaseLockQuery = `
+select count(*) from pg_locks
+-- We need to constrain this query to the correct database because we're not
+-- running two distinct Postgres instances, rather two separate databases in
+-- the same Postgres instance.
+left join pg_database on pg_locks.database = pg_database.oid -- pg_locks.database refers to the database oid, we need to check the name.
+where
+	pg_locks.locktype       = 'advisory'
+	and pg_locks.granted    = true       -- the lock must be granted, not awaited
+	and pg_locks.objid      = 3865661975 -- magic number set by the schema manager
+	and pg_database.datname = $1
+`
+
+func TestReloadControllerDatabase(t *testing.T) {
+	td, err := os.MkdirTemp("", "boundary-test-")
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, os.RemoveAll(td)) })
+
+	// Set the close time to something small
+	db.CloseSwappedDbDuration = 5 * time.Second
+
+	// Create and migrate database A and B.
+	controllerKey := config.DevKeyGeneration()
+
+	closeA, urlA, dbNameA, err := getInitDatabase(t, controllerKey)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, closeA()) })
+
+	closeB, urlB, dbNameB, err := getInitDatabase(t, controllerKey)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, closeB()) })
+
+	cmd := testServerCommand(t, testServerCommandOpts{})
+
+	workerAuthKey := config.DevKeyGeneration()
+	recoveryKey := config.DevKeyGeneration()
+	cfgHcl := fmt.Sprintf(dbSwapConfig, urlA, controllerKey, workerAuthKey, recoveryKey)
+	require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		args := []string{"-config", td + "/config.hcl"}
+		exitCode := cmd.Run(args)
+		if exitCode != 0 {
+			output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String()
+			t.Errorf("got a non-zero exit status: %s", output)
+		}
+	}()
+
+	// Wait until things are up and running (or timeout).
+	select {
+	case <-cmd.startedCh:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+
+	require.NotNil(t, cmd.schemaManager)
+
+	sqlDB, err := cmd.Server.Database.SqlDB(context.Background())
+	require.NoError(t, err)
+	require.NotNil(t, sqlDB)
+
+	// Assert we're connected to database A.
+	var currentDB string
+	row := sqlDB.QueryRow("select current_database();")
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&currentDB))
+	require.Equal(t, dbNameA, currentDB)
+
+	// Assert we've grabbed a lock on database A.
+	var lock int
+	row = sqlDB.QueryRow(getDatabaseLockQuery, dbNameA)
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&lock))
+	require.Equal(t, 1, lock)
+
+	// Get old values so we can test against them later.
+	oldSchemaManager := cmd.schemaManager
+	oldDB := cmd.Server.Database
+
+	// Change config and SIGHUP.
+	cfgHcl = fmt.Sprintf(dbSwapConfig, urlB, controllerKey, workerAuthKey, recoveryKey)
+	require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
+
+	cmd.SighupCh <- struct{}{}
+	select {
+	case <-cmd.reloadedCh:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+
+	// Assert that the schema manager ptr and value changed
+	if oldSchemaManager == cmd.schemaManager {
+		t.Fatalf("schema manager pointers are equal (%p). expected difference", oldSchemaManager)
+	}
+	require.NotEqualValues(t, oldSchemaManager, cmd.schemaManager)
+
+	// Assert that the db.DB hasn't changed.
+	// It looks like everything is the same because we're not
+	// replacing `Server.Database`, and we're not changing the
+	// underlying pointer to *dbw.DB either.
+	// We're actually replacing the value of `wrapped` (*dbw.DB)
+	// in-place (without changing memory addr).
+	// Since we can't access `db.DB.wrapped` here, that has to be
+	// tested separately on the appropriate package.
+	if oldDB != cmd.Server.Database {
+		t.Fatalf("server *db.DB pointers differ, expected equal. old ptr %p | new ptr %p", oldDB, cmd.Server.Database)
+	}
+	require.EqualValues(t, oldDB, cmd.Server.Database)
+
+	// Wait for the old connection to be closed
+	time.Sleep(db.CloseSwappedDbDuration)
+
+	// `sqlDB` still points to database A here. Assert that the object
+	// is Closed.
+	row = sqlDB.QueryRow("select 1")
+	require.ErrorContains(t, row.Err(), "database is closed")
+
+	// Get underlying *sql.DB again. We only need to do this on the test
+	// because we're getting a pointer to a *sql.DB to call Query directly
+	// and we swap the database at a higher level.
+	// At this point `sqlDB` is pointing to a memory address containing
+	// the *sql.DB used for db A, so we need to call the function again
+	// to update that reference.
+	sqlDB, err = cmd.Server.Database.SqlDB(context.Background())
+	require.NoError(t, err)
+	require.NotNil(t, sqlDB)
+
+	// Assert we're connected to database B.
+	row = sqlDB.QueryRow("select current_database();")
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&currentDB))
+	require.Equal(t, dbNameB, currentDB)
+
+	// Assert the lock on database A has been released.
+	row = sqlDB.QueryRow(getDatabaseLockQuery, dbNameA)
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&lock))
+	require.Equal(t, 0, lock)
+
+	// Assert we've grabbed a lock on database B.
+	row = sqlDB.QueryRow(getDatabaseLockQuery, dbNameB)
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&lock))
+	require.Equal(t, 1, lock)
+
+	close(cmd.ShutdownCh)
+	wg.Wait()
+}
+
+func TestReloadControllerDatabase_InvalidNewDatabaseState(t *testing.T) {
+	td, err := os.MkdirTemp("", "boundary-test-")
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, os.RemoveAll(td)) })
+
+	// Create and migrate database A and B.
+	controllerKey := config.DevKeyGeneration()
+
+	closeA, urlA, dbNameA, err := getInitDatabase(t, controllerKey)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, closeA()) })
+
+	invalidDatabaseClose, invalidDatabaseUrl, _, err := dbtest.StartUsingTemplate("postgres") // no kms set-up.
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, invalidDatabaseClose()) })
+
+	cmd := testServerCommand(t, testServerCommandOpts{})
+
+	workerAuthKey := config.DevKeyGeneration()
+	recoveryKey := config.DevKeyGeneration()
+	cfgHcl := fmt.Sprintf(dbSwapConfig, urlA, controllerKey, workerAuthKey, recoveryKey)
+	require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		args := []string{"-config", td + "/config.hcl"}
+		exitCode := cmd.Run(args)
+		if exitCode != 0 {
+			output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String()
+			t.Errorf("got a non-zero exit status: %s", output)
+		}
+	}()
+
+	// Wait until things are up and running (or timeout).
+	select {
+	case <-cmd.startedCh:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+
+	require.NotNil(t, cmd.schemaManager)
+
+	sqlDB, err := cmd.Server.Database.SqlDB(context.Background())
+	require.NoError(t, err)
+	require.NotNil(t, sqlDB)
+
+	// Assert we're connected to database A.
+	var currentDB string
+	row := sqlDB.QueryRow("select current_database();")
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&currentDB))
+	require.Equal(t, dbNameA, currentDB)
+
+	// Assert we've grabbed a lock on database A.
+	var lock int
+	row = sqlDB.QueryRow(getDatabaseLockQuery, dbNameA)
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&lock))
+	require.Equal(t, 1, lock)
+
+	// Get old values so we can test against them later.
+	oldSchemaManager := cmd.schemaManager
+
+	// Change config and SIGHUP.
+	cfgHcl = fmt.Sprintf(dbSwapConfig, invalidDatabaseUrl, controllerKey, workerAuthKey, recoveryKey)
+	require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
+
+	cmd.SighupCh <- struct{}{}
+	select {
+	case <-cmd.reloadedCh:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+
+	// Assert that the schema manager ptr and value did not change.
+	if oldSchemaManager != cmd.schemaManager {
+		t.Fatalf("schema manager pointers are different (old: %p / new: %p). expected equal", oldSchemaManager, cmd.schemaManager)
+	}
+	require.EqualValues(t, oldSchemaManager, cmd.schemaManager)
+
+	// Assert we're still connected to and locked on database A.
+	sqlDB, err = cmd.Server.Database.SqlDB(context.Background())
+	require.NoError(t, err)
+	require.NotNil(t, sqlDB)
+
+	row = sqlDB.QueryRow("select current_database();")
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&currentDB))
+	require.Equal(t, dbNameA, currentDB)
+
+	row = sqlDB.QueryRow(getDatabaseLockQuery, dbNameA)
+	require.NoError(t, row.Err())
+	require.NoError(t, row.Scan(&lock))
+	require.Equal(t, 1, lock)
+
+	close(cmd.ShutdownCh)
+	wg.Wait()
+}
+
+func TestReloadControllerDatabase_VariousNilValues(t *testing.T) {
+	// There's not much we can test in these cases, however
+	// we can ensure things don't panic.
+	tests := []struct {
+		name      string
+		cmd       *Command
+		newConfig *config.Config
+	}{
+		{
+			name: "nilServer",
+			cmd: &Command{
+				Server:     nil,
+				controller: &controller.Controller{},
+				Config: &config.Config{
+					Controller: &config.Controller{
+						Database: &config.Database{Url: "db_url"},
+					},
+				},
+			},
+			newConfig: &config.Config{
+				Controller: &config.Controller{
+					Database: &config.Database{Url: "db_url_new"},
+				},
+			},
+		},
+		{
+			name: "nilServerDatabase",
+			cmd: &Command{
+				Server:     &base.Server{Database: nil},
+				controller: &controller.Controller{},
+				Config: &config.Config{
+					Controller: &config.Controller{
+						Database: &config.Database{Url: "db_url"},
+					},
+				},
+			},
+			newConfig: &config.Config{
+				Controller: &config.Controller{
+					Database: &config.Database{Url: "db_url_new"},
+				},
+			},
+		},
+		{
+			name: "nilController",
+			cmd: &Command{
+				Server:     &base.Server{Database: &db.DB{}},
+				controller: nil,
+				Config: &config.Config{
+					Controller: &config.Controller{
+						Database: &config.Database{Url: "db_url"},
+					},
+				},
+			},
+			newConfig: &config.Config{
+				Controller: &config.Controller{
+					Database: &config.Database{Url: "db_url_new"},
+				},
+			},
+		},
+		{
+			name: "nilNewConfig",
+			cmd: &Command{
+				Server:     &base.Server{Database: &db.DB{}},
+				controller: &controller.Controller{},
+				Config: &config.Config{
+					Controller: &config.Controller{
+						Database: &config.Database{Url: "db_url"},
+					},
+				},
+			},
+			newConfig: nil,
+		},
+		{
+			name: "nilNewConfigController",
+			cmd: &Command{
+				Server:     &base.Server{Database: &db.DB{}},
+				controller: &controller.Controller{},
+				Config: &config.Config{
+					Controller: &config.Controller{
+						Database: &config.Database{Url: "db_url"},
+					},
+				},
+			},
+			newConfig: &config.Config{Controller: nil},
+		},
+		{
+			name: "nilNewConfigControllerDatabase",
+			cmd: &Command{
+				Server:     &base.Server{Database: &db.DB{}},
+				controller: &controller.Controller{},
+				Config: &config.Config{
+					Controller: &config.Controller{
+						Database: &config.Database{Url: "db_url"},
+					},
+				},
+			},
+			newConfig: &config.Config{
+				Controller: &config.Controller{Database: nil},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.NotPanics(t, func() {
+				require.NoError(t, tt.cmd.reloadControllerDatabase(tt.newConfig))
+			})
+		})
+	}
+}
+
+// getInitDatabase creates a database and sets the KMS up.
+func getInitDatabase(t *testing.T, kmsRootKey string) (func() error, string, string, error) {
+	close, url, dbName, err := dbtest.StartUsingTemplate("postgres")
+	require.NoError(t, err)
+
+	cmd := testServerCommand(t, testServerCommandOpts{})
+	cmd.Server.DatabaseUrl = url
+
+	kmsConfig := fmt.Sprintf(`
+	kms "aead" {
+		purpose = "root"
+		aead_type = "aes-gcm"
+		key = "%s"
+		key_id = "global_root"
+	}`, kmsRootKey)
+
+	cfg, err := config.Parse(kmsConfig)
+	require.NoError(t, err)
+	require.NoError(t, cmd.SetupKMSes(context.Background(), cli.NewMockUi(), cfg))
+
+	require.NoError(t, cmd.Server.OpenAndSetServerDatabase(context.Background(), "postgres"))
+	require.NoError(t, cmd.Server.CreateGlobalKmsKeys(context.Background()))
+
+	return close, url, dbName, err
+}

internal/cmd/commands/server/listener_reload_test.go

@@ -12,7 +12,6 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"sync"
 	"testing"
@@ -88,7 +87,7 @@ func TestServer_ReloadListener(t *testing.T) {
 	wd, _ := os.Getwd()
 	wd += "/test-fixtures/reload/"
 
-	td, err := ioutil.TempDir("", "boundary-test-")
+	td, err := os.MkdirTemp("", "boundary-test-")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -113,15 +112,15 @@ func TestServer_ReloadListener(t *testing.T) {
 	require.NoError(err)
 
 	// Setup initial certs
-	inBytes, err := ioutil.ReadFile(wd + "bundle1.pem")
+	inBytes, err := os.ReadFile(wd + "bundle1.pem")
 	require.NoError(err)
-	require.NoError(ioutil.WriteFile(td+"/bundle.pem", inBytes, 0o777))
+	require.NoError(os.WriteFile(td+"/bundle.pem", inBytes, 0o777))
 
 	relHcl := fmt.Sprintf(reloadConfig, cmd.DatabaseUrl, controllerKey, workerAuthKey, recoveryKey, td, td)
-	require.NoError(ioutil.WriteFile(td+"/reload.hcl", []byte(relHcl), 0o777))
+	require.NoError(os.WriteFile(td+"/reload.hcl", []byte(relHcl), 0o777))
 
 	// Populate CA pool
-	inBytes, _ = ioutil.ReadFile(td + "/bundle.pem")
+	inBytes, _ = os.ReadFile(td + "/bundle.pem")
 	certPool := x509.NewCertPool()
 	require.True(certPool.AppendCertsFromPEM(inBytes))
 
@@ -155,9 +154,9 @@ func TestServer_ReloadListener(t *testing.T) {
 
 	testCertificateSerial("142541707881583626546634262782315760343015820827")
 
-	inBytes, err = ioutil.ReadFile(wd + "bundle2.pem")
+	inBytes, err = os.ReadFile(wd + "bundle2.pem")
 	require.NoError(err)
-	require.NoError(ioutil.WriteFile(td+"/bundle.pem", inBytes, 0o777))
+	require.NoError(os.WriteFile(td+"/bundle.pem", inBytes, 0o777))
 
 	cmd.SighupCh <- struct{}{}
 	select {

internal/cmd/commands/server/server.go

@@ -50,9 +50,11 @@ type Command struct {
 	SighupCh  chan struct{}
 	SigUSR2Ch chan struct{}
 
-	Config     *config.Config
-	controller *controller.Controller
-	worker     *worker.Worker
+	Config *config.Config
+
+	schemaManager *schema.Manager
+	controller    *controller.Controller
+	worker        *worker.Worker
 
 	flagConfig      string
 	flagConfigKms   string
@@ -395,63 +397,48 @@ func (c *Command) Run(args []string) int {
 		c.DatabaseMaxIdleConnections = c.Config.Controller.Database.MaxIdleConnections
 		c.DatabaseConnMaxIdleTimeDuration = c.Config.Controller.Database.ConnMaxIdleTimeDuration
 
-		if err := c.ConnectToDatabase(c.Context, "postgres"); err != nil {
+		if err := c.OpenAndSetServerDatabase(c.Context, "postgres"); err != nil {
 			c.UI.Error(fmt.Errorf("Error connecting to database: %w", err).Error())
 			return base.CommandCliError
 		}
 
-		underlyingDB, err := c.Database.SqlDB(c.Context)
+		sm, err := acquireSchemaManager(c.Context, c.Server.Database, c.Config.Controller.Database.SkipSharedLockAcquisition)
 		if err != nil {
-			c.UI.Error(fmt.Errorf("Can't get db: %w.", err).Error())
+			c.UI.Error(fmt.Errorf("Failed to acquire database shared lock: %w", err).Error())
 			return base.CommandCliError
 		}
-		sMan, err := schema.NewManager(c.Context, "postgres", underlyingDB)
+		c.schemaManager = sm
+
+		defer func() {
+			if c.schemaManager == nil {
+				c.UI.Error("no schema manager to unlock database with")
+				return
+			}
+
+			// The base context has already been canceled so we shouldn't use it here.
+			// 1 second is chosen so the shutdown is still responsive and this is a mostly
+			// non critical step since the lock should be released when the session with the
+			// database is closed.
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+			defer cancel()
+
+			err := c.schemaManager.Close(ctx)
+			if err != nil {
+				c.UI.Error(fmt.Errorf("Unable to release shared lock to the database: %w", err).Error())
+			}
+		}()
+
+		err = verifyDatabaseState(c.Context, c.Server.Database, c.schemaManager)
 		if err != nil {
-			c.UI.Error(fmt.Errorf("Can't get schema manager: %w.", err).Error())
+			c.UI.Error(err.Error())
 			return base.CommandCliError
 		}
-		if !c.Config.Controller.Database.SkipSharedLockAcquisition {
-			// This is an advisory lock on the DB which is released when the db session ends.
-			if err := sMan.SharedLock(c.Context); err != nil {
-				c.UI.Error(fmt.Errorf("Unable to gain shared access to the database: %w", err).Error())
+
+		if c.Config.Controller.Database.SkipSharedLockAcquisition {
+			if err := c.schemaManager.Close(c.Context); err != nil {
+				c.UI.Error(fmt.Errorf("Unable to release shared lock to the database: %w", err).Error())
 				return base.CommandCliError
 			}
-			defer func() {
-				// The base context has already been canceled so we shouldn't use it here.
-				// 1 second is chosen so the shutdown is still responsive and this is a mostly
-				// non critical step since the lock should be released when the session with the
-				// database is closed.
-				ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
-				defer cancel()
-				if err := sMan.SharedUnlock(ctx); err != nil {
-					c.UI.Error(fmt.Errorf("Unable to release shared lock to the database: %w", err).Error())
-				}
-			}()
-		}
-		ckState, err := sMan.CurrentState(c.Context)
-		if err != nil {
-			c.UI.Error(fmt.Errorf("Error checking schema state: %w", err).Error())
-			return base.CommandCliError
-		}
-		if !ckState.Initialized {
-			c.UI.Error(base.WrapAtLength("The database has not been initialized. Please run 'boundary database init'."))
-			return base.CommandCliError
-		}
-		if !ckState.MigrationsApplied() {
-			for _, e := range ckState.Editions {
-				if e.DatabaseSchemaState == schema.Ahead {
-					c.UI.Error(base.WrapAtLength(fmt.Sprintf("Newer schema version (%s %d) "+
-						"than this binary expects. Please use a newer version of the boundary "+
-						"binary.", e.Name, e.DatabaseSchemaVersion)))
-					return base.CommandCliError
-				}
-			}
-			c.UI.Error(base.WrapAtLength("Database schema must be updated to use this version. Run 'boundary database migrate' to update the database. NOTE: Boundary does not currently support live migration; ensure all controllers are shut down before running the migration command."))
-			return base.CommandCliError
-		}
-		if err := c.verifyKmsSetup(); err != nil {
-			c.UI.Error(base.WrapAtLength("Database is in a bad state. Please revert the database into the last known good state."))
-			return base.CommandCliError
 		}
 	}
 
@@ -817,6 +804,11 @@ func (c *Command) Reload(newConf *config.Config) error {
 		}
 	}
 
+	err := c.reloadControllerDatabase(newConf)
+	if err != nil {
+		reloadErrors = multierror.Append(reloadErrors, fmt.Errorf("failed to reload controller database: %w", err))
+	}
+
 	if newConf != nil && c.worker != nil {
 		workerReloadErr := func() error {
 			if newConf.Controller != nil {
@@ -848,9 +840,9 @@ func (c *Command) Reload(newConf *config.Config) error {
 	return reloadErrors.ErrorOrNil()
 }
 
-func (c *Command) verifyKmsSetup() error {
+func verifyKmsSetup(dbase *db.DB) error {
 	const op = "server.(Command).verifyKmsExists"
-	rw := db.New(c.Database)
+	rw := db.New(dbase)
 
 	ctx := context.Background()
 	kmsCache, err := kms.New(ctx, rw, rw)
@@ -863,6 +855,141 @@ func (c *Command) verifyKmsSetup() error {
 	return nil
 }
 
+func (c *Command) reloadControllerDatabase(newConfig *config.Config) error {
+	if c.Server == nil || c.Server.Database == nil {
+		return nil
+	}
+	if c.controller == nil {
+		return nil
+	}
+	if newConfig == nil || newConfig.Controller == nil || newConfig.Controller.Database == nil {
+		return nil
+	}
+
+	var err error
+	newConfig.Controller.Database.Url, err = parseutil.ParsePath(newConfig.Controller.Database.Url)
+	if err != nil && !errors.Is(err, parseutil.ErrNotAUrl) {
+		return fmt.Errorf("failed to parse db url: %w", err)
+	}
+	if len(newConfig.Controller.Database.Url) == 0 || c.DatabaseUrl == newConfig.Controller.Database.Url {
+		return nil
+	}
+
+	newDb, err := c.Server.OpenDatabase(c.Context, "postgres", newConfig.Controller.Database.Url)
+	if err != nil {
+		return fmt.Errorf("failed to open connection to new database: %w", err)
+	}
+
+	// Acquire new lock on the new database and verify that it's in a good state to be used.
+	newDbSchemaManager, err := acquireSchemaManager(c.Context, newDb, c.Config.Controller.Database.SkipSharedLockAcquisition)
+	if err != nil {
+		_ = newDb.Close(c.Context)
+		return fmt.Errorf("failed to acquire shared lock on new database: %w", err)
+	}
+
+	err = verifyDatabaseState(c.Context, newDb, newDbSchemaManager)
+	if err != nil {
+		_ = newDbSchemaManager.Close(c.Context)
+		_ = newDb.Close(c.Context)
+		return fmt.Errorf("invalid new database state: %w", err)
+	}
+
+	if newConfig.Controller.Database.SkipSharedLockAcquisition {
+		if err := newDbSchemaManager.Close(c.Context); err != nil {
+			return fmt.Errorf("unable to release shared lock to the database for new schema manager: %w", err)
+		}
+	}
+
+	oldDbSchemaManager := c.schemaManager
+
+	// Swap underlying database with new one and update application state.
+	oldDbCloseFn, err := c.Database.Swap(c.Context, newDb)
+	if err != nil {
+		_ = newDbSchemaManager.Close(c.Context)
+		_ = newDb.Close(c.Context)
+		return fmt.Errorf("failed to swap databases: %w", err)
+	}
+	c.schemaManager = newDbSchemaManager
+	c.Server.DatabaseUrl = newConfig.Controller.Database.Url
+	c.Config.Controller.Database.Url = newConfig.Controller.Database.Url
+
+	// Release old database shared lock and close old database object.
+	_ = oldDbSchemaManager.Close(c.Context)
+	oldDbCloseFn(c.Context)
+
+	return nil
+}
+
+// acquireSchemaManager returns a schema manager and generally acquires a shared lock on
+// the database. This is done as a mechanism to disallow running migration commands
+// while the database is in use.
+func acquireSchemaManager(ctx context.Context, db *db.DB, skipSharedLock bool) (*schema.Manager, error) {
+	if db == nil {
+		return nil, fmt.Errorf("nil database")
+	}
+
+	underlyingDb, err := db.SqlDB(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to obtain sql db: %w", err)
+	}
+
+	manager, err := schema.NewManager(ctx, "postgres", underlyingDb)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create new schema manager: %w", err)
+	}
+
+	// This is an advisory locks on the DB which is released when the db session ends.
+	if !skipSharedLock {
+		err = manager.SharedLock(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("failed to gain shared access to the database: %w", err)
+		}
+	}
+
+	return manager, nil
+}
+
+// verifyDatabaseState checks that the migrations and kms setup for the given database are correctly setup.
+func verifyDatabaseState(ctx context.Context, db *db.DB, schemaManager *schema.Manager) error {
+	if db == nil {
+		return fmt.Errorf("nil database")
+	}
+	if schemaManager == nil {
+		return fmt.Errorf("nil schema manager")
+	}
+
+	s, err := schemaManager.CurrentState(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get current schema state: %w", err)
+	}
+	if !s.Initialized {
+		return fmt.Errorf("The database has not been initialized. Please ensure your database user " +
+			"has access to all Boundary tables, or run 'boundary database init' if you haven't initialized " +
+			"your database for Boundary.")
+	}
+	if !s.MigrationsApplied() {
+		for _, e := range s.Editions {
+			if e.DatabaseSchemaState == schema.Ahead {
+				return fmt.Errorf("Newer schema version (%s %d) "+
+					"than this binary expects. Please use a newer version of the boundary "+
+					"binary.", e.Name, e.DatabaseSchemaVersion)
+			}
+		}
+		return fmt.Errorf("Database schema must be updated to use this version. " +
+			"Run 'boundary database migrate' to update the database. " +
+			"NOTE: Boundary does not currently support live migration; " +
+			"Ensure all controllers are shut down before running the migration command.")
+	}
+
+	err = verifyKmsSetup(db)
+	if err != nil {
+		return fmt.Errorf("Database is in a bad state. Please revert the database "+
+			"into the last known good state. (Failed to verify kms setup: %w)", err)
+	}
+
+	return nil
+}
+
 var extraConfigValidationFunc = func(cfg *config.Config) error {
 	if cfg.Controller == nil && cfg.Worker == nil {
 		return stderrors.New("Neither worker nor controller specified in configuration file.")

internal/daemon/controller/testing.go

@@ -664,7 +664,7 @@ func TestControllerConfig(t testing.TB, ctx context.Context, tc *TestController,
 		if _, err := schema.MigrateStore(ctx, "postgres", tc.b.DatabaseUrl); err != nil {
 			t.Fatal(err)
 		}
-		if err := tc.b.ConnectToDatabase(tc.ctx, "postgres"); err != nil {
+		if err := tc.b.OpenAndSetServerDatabase(tc.ctx, "postgres"); err != nil {
 			t.Fatal(err)
 		}
 		if !opts.DisableKmsKeyCreation {

internal/db/changesafe_reader_writer.go

@@ -0,0 +1,105 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/hashicorp/go-dbw"
+)
+
+var (
+	_ dbw.Reader = (*changeSafeDbwReader)(nil)
+	_ dbw.Writer = (*changeSafeDbwWriter)(nil)
+)
+
+// changeSafeDbwReader is a type that wraps a *db.Db as a dbw.Reader and ensures
+// that uses of the underlying database follows changes to the connection.
+type changeSafeDbwReader struct {
+	db *Db
+}
+
+func NewChangeSafeDbwReader(underlying *Db) dbw.Reader {
+	return &changeSafeDbwReader{db: underlying}
+}
+
+func (r *changeSafeDbwReader) LookupBy(ctx context.Context, resource any, opt ...dbw.Option) error {
+	return dbw.New(r.db.underlying.wrapped.Load()).LookupBy(ctx, resource, opt...)
+}
+
+func (r *changeSafeDbwReader) LookupByPublicId(ctx context.Context, resource dbw.ResourcePublicIder, opt ...dbw.Option) error {
+	return dbw.New(r.db.underlying.wrapped.Load()).LookupByPublicId(ctx, resource, opt...)
+}
+
+func (r *changeSafeDbwReader) LookupWhere(ctx context.Context, resource any, where string, args []any, opt ...dbw.Option) error {
+	return dbw.New(r.db.underlying.wrapped.Load()).LookupWhere(ctx, resource, where, args, opt...)
+}
+
+func (r *changeSafeDbwReader) Query(ctx context.Context, sql string, values []any, opt ...dbw.Option) (*sql.Rows, error) {
+	return dbw.New(r.db.underlying.wrapped.Load()).Query(ctx, sql, values, opt...)
+}
+
+func (r *changeSafeDbwReader) ScanRows(rows *sql.Rows, result any) error {
+	return dbw.New(r.db.underlying.wrapped.Load()).ScanRows(rows, result)
+}
+
+func (r *changeSafeDbwReader) SearchWhere(ctx context.Context, resources any, where string, args []any, opt ...dbw.Option) error {
+	return dbw.New(r.db.underlying.wrapped.Load()).SearchWhere(ctx, resources, where, args, opt...)
+}
+
+// changeSafeDbwWriter is a type that wraps a *db.Db as a dbw.Writer and ensures
+// that uses of the underlying database follows changes to the connection.
+type changeSafeDbwWriter struct {
+	db *Db
+}
+
+func NewChangeSafeDbwWriter(underlying *Db) dbw.Writer {
+	return &changeSafeDbwWriter{db: underlying}
+}
+
+func (w *changeSafeDbwWriter) Begin(ctx context.Context) (*dbw.RW, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).Begin(ctx)
+}
+
+func (w *changeSafeDbwWriter) Commit(ctx context.Context) error {
+	return dbw.New(w.db.underlying.wrapped.Load()).Commit(ctx)
+}
+
+func (w *changeSafeDbwWriter) Create(ctx context.Context, i any, opt ...dbw.Option) error {
+	return dbw.New(w.db.underlying.wrapped.Load()).Create(ctx, i, opt...)
+}
+
+func (w *changeSafeDbwWriter) CreateItems(ctx context.Context, createItems []any, opt ...dbw.Option) error {
+	return dbw.New(w.db.underlying.wrapped.Load()).CreateItems(ctx, createItems, opt...)
+}
+
+func (w *changeSafeDbwWriter) Delete(ctx context.Context, i any, opt ...dbw.Option) (int, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).Delete(ctx, i, opt...)
+}
+
+func (w *changeSafeDbwWriter) DeleteItems(ctx context.Context, deleteItems []any, opt ...dbw.Option) (int, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).DeleteItems(ctx, deleteItems, opt...)
+}
+
+func (w *changeSafeDbwWriter) DoTx(ctx context.Context, retryErrorsMatchingFn func(error) bool, retries uint, backOff dbw.Backoff, handler dbw.TxHandler) (dbw.RetryInfo, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).DoTx(ctx, retryErrorsMatchingFn, retries, backOff, handler)
+}
+
+func (w *changeSafeDbwWriter) Exec(ctx context.Context, sql string, values []any, opt ...dbw.Option) (int, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).Exec(ctx, sql, values, opt...)
+}
+
+func (w *changeSafeDbwWriter) Query(ctx context.Context, sql string, values []any, opt ...dbw.Option) (*sql.Rows, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).Query(ctx, sql, values, opt...)
+}
+
+func (w *changeSafeDbwWriter) Rollback(ctx context.Context) error {
+	return dbw.New(w.db.underlying.wrapped.Load()).Rollback(ctx)
+}
+
+func (w *changeSafeDbwWriter) ScanRows(rows *sql.Rows, result any) error {
+	return dbw.New(w.db.underlying.wrapped.Load()).ScanRows(rows, result)
+}
+
+func (w *changeSafeDbwWriter) Update(ctx context.Context, i interface{}, fieldMaskPaths []string, setToNullPaths []string, opt ...dbw.Option) (int, error) {
+	return dbw.New(w.db.underlying.wrapped.Load()).Update(ctx, i, fieldMaskPaths, setToNullPaths, opt...)
+}

internal/db/db.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"sync/atomic"
+	"time"
 
 	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-dbw"
 	_ "github.com/jackc/pgx/v4"
 	"gorm.io/driver/postgres"
@@ -41,12 +44,68 @@ func StringToDbType(dialect string) (DbType, error) {
 
 // DB is a wrapper around the ORM
 type DB struct {
-	wrapped *dbw.DB
+	wrapped *atomic.Pointer[dbw.DB]
+}
+
+type closeDbFn func(context.Context)
+
+var CloseSwappedDbDuration = 5 * time.Minute
+
+// Swap replaces *DB's underlying database object with the one from `newDB`.
+// It returns a function that calls Close() on the outgoing database object.
+// Note: Swap does not verify the incoming *DB object is correctly set-up.
+func (db *DB) Swap(ctx context.Context, newDB *DB) (closeDbFn, error) {
+	const op = "db.(DB).Swap"
+	if newDB == nil || newDB.wrapped == nil || newDB.wrapped.Load() == nil {
+		return nil, fmt.Errorf("no new db object present")
+	}
+	if db == nil || db.wrapped == nil || db.wrapped.Load() == nil {
+		// TBD: We could be helpful here and set the new DB instead of err?
+		return nil, fmt.Errorf("no current db is present to swap, aborting")
+	}
+
+	// Grab the old db to allow for cleanup after swap.
+	oldDbw := db.wrapped.Swap(newDB.wrapped.Load())
+	closeOldDbFn := func(ctx context.Context) {
+		go func() {
+			maxTime := time.Now().Add(CloseSwappedDbDuration)
+			t := time.NewTicker(time.Second)
+			var done bool
+			for {
+				select {
+				case <-t.C:
+					if time.Now().After(maxTime) || done {
+						t.Stop()
+						if err := oldDbw.Close(ctx); err != nil {
+							event.WriteError(ctx, op, errors.Wrap(ctx, err, op))
+						}
+						return
+					}
+					sqlDb, err := oldDbw.SqlDB(ctx)
+					if err != nil {
+						event.WriteError(ctx, op, fmt.Errorf("unable to load old sqldb to check stats"))
+						continue
+					}
+					stats := sqlDb.Stats()
+					if stats.InUse == 0 {
+						done = true
+					}
+
+				case <-ctx.Done():
+					t.Stop()
+					event.WriteError(ctx, op, fmt.Errorf("context canceled before old database connection was closed, aborting"))
+					return
+				}
+			}
+		}()
+	}
+
+	return closeOldDbFn, nil
 }
 
 // Debug will enable/disable debug info for the connection
 func (db *DB) Debug(on bool) {
-	db.wrapped.Debug(on)
+	db.wrapped.Load().Debug(on)
 }
 
 // SqlDB returns the underlying sql.DB
@@ -59,7 +118,7 @@ func (d *DB) SqlDB(ctx context.Context) (*sql.DB, error) {
 	if d.wrapped == nil {
 		return nil, errors.New(ctx, errors.Internal, op, "missing underlying database")
 	}
-	return d.wrapped.SqlDB(ctx)
+	return d.wrapped.Load().SqlDB(ctx)
 }
 
 // Close the underlying sql.DB
@@ -68,7 +127,7 @@ func (d *DB) Close(ctx context.Context) error {
 	if d.wrapped == nil {
 		return errors.New(ctx, errors.Internal, op, "missing underlying database")
 	}
-	return d.wrapped.Close(ctx)
+	return d.wrapped.Load().Close(ctx)
 }
 
 // Open a database connection which is long-lived. The options of
@@ -119,5 +178,7 @@ func Open(ctx context.Context, dbType DbType, connectionUrl string, opt ...Optio
 		sdb.SetConnMaxIdleTime(*opts.withConnMaxIdleTimeDuration)
 	}
 
-	return &DB{wrapped: wrapped}, nil
+	ret := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+	ret.wrapped.Store(wrapped)
+	return ret, nil
 }

internal/db/db_test.go

@@ -2,9 +2,11 @@ package db
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 
 	"github.com/hashicorp/boundary/testing/dbtest"
+	"github.com/hashicorp/go-dbw"
 	"github.com/stretchr/testify/require"
 )
 
@@ -76,3 +78,99 @@ func TestOpen(t *testing.T) {
 		})
 	}
 }
+
+func TestSwap(t *testing.T) {
+	tests := []struct {
+		name      string
+		db        func() *DB
+		newDB     func() *DB
+		expErr    bool
+		expErrStr string
+	}{
+		{
+			name: "nilNewDB",
+			db: func() *DB {
+				ret := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+				return ret
+			},
+			newDB:     nil,
+			expErr:    true,
+			expErrStr: "no new db object present",
+		},
+		{
+			name: "nilNewDBWrapped",
+			db: func() *DB {
+				ret := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+				return ret
+			},
+			newDB: func() *DB {
+				ret := &DB{}
+				return ret
+			},
+			expErr:    true,
+			expErrStr: "no new db object present",
+		},
+		{
+			name: "nilCurrentDBWrapped",
+			db: func() *DB {
+				ret := &DB{}
+				return ret
+			},
+			newDB: func() *DB {
+				db, _ := dbw.TestSetupWithMock(t)
+				ret := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+				ret.wrapped.Store(db)
+				return ret
+			},
+			expErr:    true,
+			expErrStr: "no current db is present to swap, aborting",
+		},
+		{
+			name: "dbReplace",
+			db: func() *DB {
+				db, _ := dbw.TestSetupWithMock(t)
+				ret := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+				ret.wrapped.Store(db)
+				return ret
+			},
+			newDB: func() *DB {
+				db, _ := dbw.TestSetupWithMock(t)
+				ret := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+				ret.wrapped.Store(db)
+				return ret
+			},
+			expErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var oldDb, newDb *DB
+			if tt.db != nil {
+				oldDb = tt.db()
+			}
+			if tt.newDB != nil {
+				newDb = tt.newDB()
+			}
+			var oldWrappedVal dbw.DB
+			if !tt.expErr {
+				oldWrappedVal = *oldDb.wrapped.Load()
+			}
+
+			ctx := context.Background()
+			closeFn, err := oldDb.Swap(ctx, newDb)
+			if tt.expErr {
+				require.EqualError(t, err, tt.expErrStr)
+				require.Nil(t, closeFn)
+				return
+			}
+
+			dbw.TestSetupWithMock(t)
+			require.NoError(t, err)
+			require.NotNil(t, closeFn)
+
+			require.NotEqual(t, oldWrappedVal, oldDb.wrapped.Load())
+			require.EqualValues(t, newDb.wrapped.Load(), oldDb.wrapped.Load()) // For pointer values, require tests the underlying values' equality.
+		})
+	}
+}

internal/db/read_writer.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"fmt"
 	"reflect"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/errors"
@@ -208,9 +209,13 @@ func New(underlying *DB) *Db {
 	return &Db{underlying: underlying}
 }
 
-// UnderlyingDB returns the underlying *dbw.DB
-func (rw *Db) UnderlyingDB() *dbw.DB {
-	return rw.underlying.wrapped
+// UnderlyingDB returns a function to get the underlying *dbw.DB. The function
+// should be called every time rather than caching the value, as the value may
+// change from call to call.
+func (rw *Db) UnderlyingDB() func() *dbw.DB {
+	return func() *dbw.DB {
+		return rw.underlying.wrapped.Load()
+	}
 }
 
 // Exec will execute the sql with the values as parameters. The int returned
@@ -221,7 +226,7 @@ func (rw *Db) Exec(ctx context.Context, sql string, values []interface{}, opt ..
 		return NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "missing sql")
 	}
 	opts := GetOpts(opt...)
-	rowsAffected, err := dbw.New(rw.underlying.wrapped).Exec(ctx, sql, values, dbw.WithDebug(opts.withDebug))
+	rowsAffected, err := dbw.New(rw.underlying.wrapped.Load()).Exec(ctx, sql, values, dbw.WithDebug(opts.withDebug))
 	if err != nil {
 		return NoRowsAffected, wrapError(ctx, err, op)
 	}
@@ -238,7 +243,7 @@ func (rw *Db) Query(ctx context.Context, sql string, values []interface{}, opt .
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing sql")
 	}
 	opts := GetOpts(opt...)
-	rows, err := dbw.New(rw.underlying.wrapped).Query(ctx, sql, values, dbw.WithDebug(opts.withDebug))
+	rows, err := dbw.New(rw.underlying.wrapped.Load()).Query(ctx, sql, values, dbw.WithDebug(opts.withDebug))
 	if err != nil {
 		return nil, wrapError(ctx, err, op)
 	}
@@ -254,7 +259,7 @@ func (rw *Db) ScanRows(ctx context.Context, rows *sql.Rows, result interface{})
 	if isNil(result) {
 		return errors.New(ctx, errors.InvalidParameter, op, "missing result")
 	}
-	if err := dbw.New(rw.underlying.wrapped).ScanRows(rows, result); err != nil {
+	if err := dbw.New(rw.underlying.wrapped.Load()).ScanRows(rows, result); err != nil {
 		return wrapError(ctx, err, op)
 	}
 	return nil
@@ -284,7 +289,7 @@ func (rw *Db) Create(ctx context.Context, i interface{}, opt ...Option) error {
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
-	if err := dbw.New(rw.underlying.wrapped).Create(ctx, i, dbwOpts...); err != nil {
+	if err := dbw.New(rw.underlying.wrapped.Load()).Create(ctx, i, dbwOpts...); err != nil {
 		return wrapError(ctx, err, op)
 	}
 	return nil
@@ -303,7 +308,7 @@ func (rw *Db) CreateItems(ctx context.Context, createItems []interface{}, opt ..
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
-	if err := dbw.New(rw.underlying.wrapped).CreateItems(ctx, createItems, dbwOpts...); err != nil {
+	if err := dbw.New(rw.underlying.wrapped.Load()).CreateItems(ctx, createItems, dbwOpts...); err != nil {
 		return wrapError(ctx, err, op)
 	}
 	return nil
@@ -341,7 +346,7 @@ func (rw *Db) Update(ctx context.Context, i interface{}, fieldMaskPaths []string
 	if err != nil {
 		return NoRowsAffected, errors.Wrap(ctx, err, op)
 	}
-	rowsUpdated, err := dbw.New(rw.underlying.wrapped).Update(ctx, i, fieldMaskPaths, setToNullPaths, dbwOpts...)
+	rowsUpdated, err := dbw.New(rw.underlying.wrapped.Load()).Update(ctx, i, fieldMaskPaths, setToNullPaths, dbwOpts...)
 	if err != nil {
 		return NoRowsAffected, wrapError(ctx, err, op)
 	}
@@ -362,7 +367,7 @@ func (rw *Db) Delete(ctx context.Context, i interface{}, opt ...Option) (int, er
 	if err != nil {
 		return NoRowsAffected, wrapError(ctx, err, op)
 	}
-	rowsUpdated, err := dbw.New(rw.underlying.wrapped).Delete(ctx, i, dbwOpts...)
+	rowsUpdated, err := dbw.New(rw.underlying.wrapped.Load()).Delete(ctx, i, dbwOpts...)
 	if err != nil {
 		return NoRowsAffected, wrapError(ctx, err, op)
 	}
@@ -381,7 +386,7 @@ func (rw *Db) DeleteItems(ctx context.Context, deleteItems []interface{}, opt ..
 	if err != nil {
 		return NoRowsAffected, errors.Wrap(ctx, err, op)
 	}
-	rowsDeleted, err := dbw.New(rw.underlying.wrapped).DeleteItems(ctx, deleteItems, dbwOpts...)
+	rowsDeleted, err := dbw.New(rw.underlying.wrapped.Load()).DeleteItems(ctx, deleteItems, dbwOpts...)
 	if err != nil {
 		return NoRowsAffected, wrapError(ctx, err, op)
 	}
@@ -410,11 +415,14 @@ func (w *Db) DoTx(ctx context.Context, retries uint, backOff Backoff, handler Tx
 		}
 
 		// step one of this, start a transaction...
-		beginTx, err := dbw.New(w.underlying.wrapped).Begin(ctx)
+		beginTx, err := dbw.New(w.underlying.wrapped.Load()).Begin(ctx)
 		if err != nil {
 			return info, wrapError(ctx, err, op)
 		}
-		newRW := New(&DB{wrapped: beginTx.DB()})
+
+		newTxDb := &DB{wrapped: new(atomic.Pointer[dbw.DB])}
+		newTxDb.wrapped.Store(beginTx.DB())
+		newRW := New(newTxDb)
 
 		if err := handler(newRW, newRW); err != nil {
 			if err := beginTx.Rollback(ctx); err != nil {
@@ -448,7 +456,7 @@ func (rw *Db) LookupById(ctx context.Context, resourceWithIder interface{}, opt
 		return errors.New(ctx, errors.InvalidParameter, op, "missing underlying db")
 	}
 	opts := GetOpts(opt...)
-	if err := dbw.New(rw.underlying.wrapped).LookupBy(ctx, resourceWithIder, dbw.WithDebug(opts.withDebug)); err != nil {
+	if err := dbw.New(rw.underlying.wrapped.Load()).LookupBy(ctx, resourceWithIder, dbw.WithDebug(opts.withDebug)); err != nil {
 		var errOpts []errors.Option
 		if errors.Is(err, dbw.ErrRecordNotFound) {
 			// Not found is a common workflow in the application layer during lookup, suppress
@@ -474,7 +482,7 @@ func (rw *Db) LookupWhere(ctx context.Context, resource interface{}, where strin
 		return errors.New(ctx, errors.InvalidParameter, op, "missing underlying db")
 	}
 	opts := GetOpts(opt...)
-	if err := dbw.New(rw.underlying.wrapped).LookupWhere(ctx, resource, where, args, dbw.WithDebug(opts.withDebug)); err != nil {
+	if err := dbw.New(rw.underlying.wrapped.Load()).LookupWhere(ctx, resource, where, args, dbw.WithDebug(opts.withDebug)); err != nil {
 		var errOpts []errors.Option
 		if errors.Is(err, dbw.ErrRecordNotFound) {
 			// Not found is a common workflow in the application layer during lookup, suppress
@@ -502,7 +510,7 @@ func (rw *Db) SearchWhere(ctx context.Context, resources interface{}, where stri
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
-	if err := dbw.New(rw.underlying.wrapped).SearchWhere(ctx, resources, where, args, dbwOpts...); err != nil {
+	if err := dbw.New(rw.underlying.wrapped.Load()).SearchWhere(ctx, resources, where, args, dbwOpts...); err != nil {
 		return wrapError(ctx, err, op)
 	}
 	return nil

internal/db/read_writer_oplog.go

@@ -166,7 +166,7 @@ func (rw *Db) getTicketFor(ctx context.Context, aggregateName string) (*store.Ti
 	if rw.underlying == nil {
 		return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("%s: underlying db missing", aggregateName), errors.WithoutEvent())
 	}
-	ticketer, err := oplog.NewTicketer(ctx, rw.underlying.wrapped, oplog.WithAggregateNames(true))
+	ticketer, err := oplog.NewTicketer(ctx, rw.UnderlyingDB()(), oplog.WithAggregateNames(true))
 	if err != nil {
 		return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("%s: unable to get Ticketer", aggregateName)), errors.WithoutEvent())
 	}
@@ -246,7 +246,7 @@ func (rw *Db) addOplogForItems(ctx context.Context, opType OpType, opts Options,
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg("oplog validation failed"), errors.WithoutEvent())
 	}
-	ticketer, err := oplog.NewTicketer(ctx, rw.underlying.wrapped, oplog.WithAggregateNames(true))
+	ticketer, err := oplog.NewTicketer(ctx, rw.UnderlyingDB()(), oplog.WithAggregateNames(true))
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer"), errors.WithoutEvent())
 	}
@@ -262,7 +262,7 @@ func (rw *Db) addOplogForItems(ctx context.Context, opType OpType, opts Options,
 	}
 	if err := entry.WriteEntryWith(
 		ctx,
-		&oplog.Writer{DB: rw.underlying.wrapped},
+		&oplog.Writer{DB: rw.UnderlyingDB()()},
 		ticket,
 		oplogMsgs...,
 	); err != nil {
@@ -281,7 +281,7 @@ func (rw *Db) addOplog(ctx context.Context, opType OpType, opts Options, ticket
 	if ticket == nil {
 		return errors.New(ctx, errors.InvalidParameter, op, "missing ticket", errors.WithoutEvent())
 	}
-	ticketer, err := oplog.NewTicketer(ctx, rw.underlying.wrapped, oplog.WithAggregateNames(true))
+	ticketer, err := oplog.NewTicketer(ctx, rw.UnderlyingDB()(), oplog.WithAggregateNames(true))
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer"), errors.WithoutEvent())
 	}
@@ -301,7 +301,7 @@ func (rw *Db) addOplog(ctx context.Context, opType OpType, opts Options, ticket
 	}
 	err = entry.WriteEntryWith(
 		ctx,
-		&oplog.Writer{DB: rw.underlying.wrapped},
+		&oplog.Writer{DB: rw.UnderlyingDB()()},
 		ticket,
 		msg,
 	)
@@ -330,7 +330,7 @@ func (rw *Db) WriteOplogEntryWith(ctx context.Context, wrapper wrapping.Wrapper,
 	if len(metadata) == 0 {
 		return errors.New(ctx, errors.InvalidParameter, op, "missing metadata")
 	}
-	ticketer, err := oplog.NewTicketer(ctx, rw.underlying.wrapped, oplog.WithAggregateNames(true))
+	ticketer, err := oplog.NewTicketer(ctx, rw.UnderlyingDB()(), oplog.WithAggregateNames(true))
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer"))
 	}
@@ -347,7 +347,7 @@ func (rw *Db) WriteOplogEntryWith(ctx context.Context, wrapper wrapping.Wrapper,
 	}
 	err = entry.WriteEntryWith(
 		ctx,
-		&oplog.Writer{DB: rw.underlying.wrapped},
+		&oplog.Writer{DB: rw.UnderlyingDB()()},
 		ticket,
 		msgs...,
 	)

internal/db/schema/internal/postgres/postgres.go

@@ -251,6 +251,11 @@ func (p *Postgres) Run(ctx context.Context, migration io.Reader, version int, ed
 	return nil
 }
 
+// Close closes the underlying Postgres database connection.
+func (p *Postgres) Close() error {
+	return p.conn.Close()
+}
+
 var errOldMigrationTable = stderrors.New("old schema migration table")
 
 func (p *Postgres) schemaInitialized(ctx context.Context) (bool, error) {

internal/db/schema/manager.go

@@ -6,6 +6,7 @@ import (
 	"database/sql"
 	"fmt"
 	"io"
+	"sync"
 
 	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
 	"github.com/hashicorp/boundary/internal/db/schema/internal/log"
@@ -13,6 +14,7 @@ import (
 	"github.com/hashicorp/boundary/internal/db/schema/internal/provider"
 	"github.com/hashicorp/boundary/internal/db/schema/migration"
 	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/observability/event"
 )
 
 // driver provides functionality to a database.
@@ -22,6 +24,7 @@ type driver interface {
 	Lock(context.Context) error
 	Unlock(context.Context) error
 	UnlockShared(context.Context) error
+	Close() error
 	// StartRun begins a transaction internal to the driver.
 	StartRun(context.Context) error
 	// CommitRun commits a transaction, if there is an error it should rollback the transaction.
@@ -58,11 +61,13 @@ type driver interface {
 // the underlying boundary database schema.
 // Manager is not thread safe.
 type Manager struct {
-	db              *sql.DB
-	driver          driver
-	dialect         string
-	editions        edition.Editions
-	selectedRepairs RepairMigrations
+	db                 *sql.DB
+	driver             driver
+	dialect            string
+	editions           edition.Editions
+	selectedRepairs    RepairMigrations
+	sharedLockAcquired bool
+	sharedLockMutex    *sync.Mutex
 }
 
 // NewManager creates a new schema manager. An error is returned
@@ -78,6 +83,7 @@ func NewManager(ctx context.Context, dialect Dialect, db *sql.DB, opt ...Option)
 		db:              db,
 		dialect:         string(dialect),
 		selectedRepairs: opts.withRepairMigrations,
+		sharedLockMutex: new(sync.Mutex),
 	}
 	if opts.withEditions != nil {
 		dbM.editions = opts.withEditions
@@ -123,14 +129,38 @@ func (b *Manager) CurrentState(ctx context.Context) (*State, error) {
 	return &dbS, nil
 }
 
+// Close unlocks the database shared lock and closes the underlying
+// database connection afterwards.
+func (b *Manager) Close(ctx context.Context) error {
+	const op = "schema.(Manager).Close"
+
+	if err := b.SharedUnlock(ctx); err != nil {
+		event.WriteError(ctx, op, fmt.Errorf("error unlocking shared database lock: %w", err))
+	}
+
+	if err := b.driver.Close(); err != sql.ErrConnDone {
+		return err
+	}
+
+	return nil
+}
+
 // SharedLock attempts to obtain a shared lock on the database.  This can fail
 // if an exclusive lock is already held.  If the lock can't be obtained an
 // error is returned.
 func (b *Manager) SharedLock(ctx context.Context) error {
 	const op = "schema.(Manager).SharedLock"
-	if err := b.driver.TrySharedLock(ctx); err != nil {
-		return errors.Wrap(ctx, err, op)
+
+	b.sharedLockMutex.Lock()
+	defer b.sharedLockMutex.Unlock()
+
+	if !b.sharedLockAcquired {
+		if err := b.driver.TrySharedLock(ctx); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		b.sharedLockAcquired = true
 	}
+
 	return nil
 }
 
@@ -139,9 +169,17 @@ func (b *Manager) SharedLock(ctx context.Context) error {
 // that is not held is not an error.
 func (b *Manager) SharedUnlock(ctx context.Context) error {
 	const op = "schema.(Manager).SharedUnlock"
-	if err := b.driver.UnlockShared(ctx); err != nil {
-		return errors.Wrap(ctx, err, op)
+
+	b.sharedLockMutex.Lock()
+	defer b.sharedLockMutex.Unlock()
+
+	if b.sharedLockAcquired {
+		if err := b.driver.UnlockShared(ctx); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		b.sharedLockAcquired = false
 	}
+
 	return nil
 }
 

internal/db/testing.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -64,7 +65,7 @@ func TestSetup(t testing.TB, dialect string, opt ...TestOption) (*DB, string) {
 	}
 	switch {
 	case opts.withLogLevel != DefaultTestLogLevel:
-		db.wrapped.LogLevel(dbw.LogLevel(opts.withLogLevel))
+		db.wrapped.Load().LogLevel(dbw.LogLevel(opts.withLogLevel))
 	default:
 		var defaultLevel dbw.LogLevel
 		switch strings.ToLower(os.Getenv("DEFAULT_TEST_LOG_LEVEL")) {
@@ -79,7 +80,7 @@ func TestSetup(t testing.TB, dialect string, opt ...TestOption) (*DB, string) {
 		default:
 			defaultLevel = dbw.Silent
 		}
-		db.wrapped.LogLevel(defaultLevel)
+		db.wrapped.Load().LogLevel(defaultLevel)
 	}
 	t.Cleanup(func() {
 		sqlDB, err := db.SqlDB(ctx)
@@ -97,13 +98,15 @@ func TestSetupWithMock(t *testing.T) (*DB, sqlmock.Sqlmock) {
 	db, mock, err := sqlmock.New()
 	require.NoError(err)
 	require.NoError(err)
-	dbw, err := dbw.OpenWith(pgDriver.New(pgDriver.Config{
+	dbWith, err := dbw.OpenWith(pgDriver.New(pgDriver.Config{
 		Conn: db,
 	}))
 	require.NoError(err)
-	return &DB{
-		wrapped: dbw,
-	}, mock
+	ret := &DB{
+		wrapped: new(atomic.Pointer[dbw.DB]),
+	}
+	ret.wrapped.Store(dbWith)
+	return ret, mock
 }
 
 // TestWrapper initializes an AEAD wrapping.Wrapper for testing the oplog
@@ -148,7 +151,7 @@ func TestDeleteWhere(t testing.TB, conn *DB, i interface{}, whereClause string,
 		TableName() string
 	})
 	require.True(ok)
-	_, err := dbw.New(conn.wrapped).Exec(ctx, fmt.Sprintf(`delete from "%s" where %s`, tabler.TableName(), whereClause), args)
+	_, err := dbw.New(conn.wrapped.Load()).Exec(ctx, fmt.Sprintf(`delete from "%s" where %s`, tabler.TableName(), whereClause), args)
 	require.NoError(err)
 }
 

internal/kms/kms.go

@@ -11,7 +11,6 @@ import (
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/types/scope"
-	"github.com/hashicorp/go-dbw"
 	wrappingKms "github.com/hashicorp/go-kms-wrapping/extras/kms/v2"
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
 )
@@ -37,9 +36,7 @@ func New(ctx context.Context, reader *db.Db, writer *db.Db, _ ...Option) (*Kms,
 	}
 	purposes := stdNewKmsPurposes()
 
-	r := dbw.New(reader.UnderlyingDB())
-	w := dbw.New(writer.UnderlyingDB())
-	k, err := wrappingKms.New(r, w, purposes)
+	k, err := wrappingKms.New(db.NewChangeSafeDbwReader(reader), db.NewChangeSafeDbwWriter(writer), purposes)
 	if err != nil {
 		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error creating new in-memory kms"))
 	}
@@ -61,15 +58,15 @@ func NewUsingReaderWriter(ctx context.Context, reader db.Reader, writer db.Write
 	}
 	purposes := stdNewKmsPurposes()
 
-	r, err := convertToRW(ctx, reader)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
+	r, ok := reader.(*db.Db)
+	if !ok {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "unable to convert reader to db.Db")
 	}
-	w, err := convertToRW(ctx, writer)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
+	w, ok := writer.(*db.Db)
+	if !ok {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "unable to convert writer to db.Db")
 	}
-	k, err := wrappingKms.New(r, w, purposes)
+	k, err := wrappingKms.New(db.NewChangeSafeDbwReader(r), db.NewChangeSafeDbwWriter(w), purposes)
 	if err != nil {
 		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error creating new in-memory kms"))
 	}
@@ -79,15 +76,6 @@ func NewUsingReaderWriter(ctx context.Context, reader db.Reader, writer db.Write
 	}, nil
 }
 
-func convertToRW(ctx context.Context, r interface{}) (*dbw.RW, error) {
-	const op = "kms.convertToDb"
-	d, ok := r.(*db.Db)
-	if !ok {
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "unable to convert to db.DB")
-	}
-	return dbw.New(d.UnderlyingDB()), nil
-}
-
 // AddExternalWrappers allows setting the external keys.
 func (k *Kms) AddExternalWrappers(ctx context.Context, opt ...Option) error {
 	const op = "kms.(Kms).AddExternalWrappers"
@@ -208,15 +196,15 @@ func (k *Kms) CreateKeys(ctx context.Context, scopeId string, opt ...Option) err
 	case isNil(opts.withReader) && !isNil(opts.withWriter):
 		return errors.New(ctx, errors.InvalidParameter, op, "missing reader")
 	case !isNil(opts.withReader) && !isNil(opts.withWriter):
-		r, err := convertToRW(ctx, opts.withReader)
-		if err != nil {
-			return errors.Wrap(ctx, err, op, errors.WithMsg("unable to convert reader"))
+		r, ok := opts.withReader.(*db.Db)
+		if !ok {
+			return errors.New(ctx, errors.InvalidParameter, op, "unable to convert reader to db.Db")
 		}
-		w, err := convertToRW(ctx, opts.withWriter)
-		if err != nil {
-			return errors.Wrap(ctx, err, op, errors.WithMsg("unable to convert writer"))
+		w, ok := opts.withWriter.(*db.Db)
+		if !ok {
+			return errors.New(ctx, errors.InvalidParameter, op, "unable to convert writer to db.Db")
 		}
-		kmsOpts = append(kmsOpts, wrappingKms.WithReaderWriter(r, w))
+		kmsOpts = append(kmsOpts, wrappingKms.WithReaderWriter(db.NewChangeSafeDbwReader(r), db.NewChangeSafeDbwWriter(w)))
 	}
 	purposes := make([]wrappingKms.KeyPurpose, 0, len(ValidDekPurposes()))
 	for _, p := range ValidDekPurposes() {

internal/kms/kms_ext_test.go

@@ -229,7 +229,7 @@ func TestKms_GetWrapper(t *testing.T) {
 	conn, _ := db.TestSetup(t, "postgres")
 	rootWrapper := db.TestWrapper(t)
 	kmsCache := kms.TestKms(t, conn, rootWrapper)
-	kmsCache.CreateKeys(testCtx, "global")
+	require.NoError(t, kmsCache.CreateKeys(testCtx, "global"))
 	tests := []struct {
 		name            string
 		kms             *kms.Kms
@@ -289,7 +289,7 @@ func TestKms_CreateKeys(t *testing.T) {
 	conn, _ := db.TestSetup(t, "postgres")
 	rootWrapper := db.TestWrapper(t)
 	kmsCache := kms.TestKms(t, conn, rootWrapper)
-	kmsCache.CreateKeys(testCtx, "global")
+	require.NoError(t, kmsCache.CreateKeys(testCtx, "global"))
 	rw := db.New(conn)
 
 	tests := []struct {

internal/kms/kms_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
-	"github.com/hashicorp/go-dbw"
 	wrappingKms "github.com/hashicorp/go-kms-wrapping/extras/kms/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -55,9 +54,7 @@ func Test_New(t *testing.T) {
 					purposes = append(purposes, wrappingKms.KeyPurpose(KeyPurposeWorkerAuth.String()),
 						wrappingKms.KeyPurpose(KeyPurposeWorkerAuthStorage.String()), wrappingKms.KeyPurpose(KeyPurposeRecovery.String()))
 
-					r := dbw.New(rw.UnderlyingDB())
-					w := dbw.New(rw.UnderlyingDB())
-					wrapped, err := wrappingKms.New(r, w, purposes)
+					wrapped, err := wrappingKms.New(db.NewChangeSafeDbwReader(rw), db.NewChangeSafeDbwWriter(rw), purposes)
 					require.NoError(t, err)
 					return wrapped
 				}(),
@@ -119,7 +116,7 @@ func Test_NewUsingReaderWriter(t *testing.T) {
 			r:               &invalidReader{},
 			w:               rw,
 			wantErr:         true,
-			wantErrContains: "unable to convert to db.DB",
+			wantErrContains: "unable to convert reader to db.Db",
 			wantErrMatch:    errors.T(errors.InvalidParameter),
 		},
 		{
@@ -127,7 +124,7 @@ func Test_NewUsingReaderWriter(t *testing.T) {
 			r:               rw,
 			w:               &invalidWriter{},
 			wantErr:         true,
-			wantErrContains: "unable to convert to db.DB",
+			wantErrContains: "unable to convert writer to db.Db",
 			wantErrMatch:    errors.T(errors.InvalidParameter),
 		},
 		{
@@ -138,9 +135,7 @@ func Test_NewUsingReaderWriter(t *testing.T) {
 				reader: rw,
 				underlying: func() *wrappingKms.Kms {
 					purposes := stdNewKmsPurposes()
-					r := dbw.New(rw.UnderlyingDB())
-					w := dbw.New(rw.UnderlyingDB())
-					wrapped, err := wrappingKms.New(r, w, purposes)
+					wrapped, err := wrappingKms.New(db.NewChangeSafeDbwReader(rw), db.NewChangeSafeDbwWriter(rw), purposes)
 					require.NoError(t, err)
 					return wrapped
 				}(),