feat(server): Reload database on SIGHUP

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. In this case only a database url config
change and a SIGHUP signal to all controllers would be required.

This uses an atomic.Pointer to store wrapped DB objects. This allows
atomically swapping to the new object. A timeout is introduced 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.

This could be done with mutexes to avoid checking stats in a loop, but
that carries its own risk: if a write lock is attempted to be acquired
while a query is holding a read lock, the write lock will block, and any
subsequent read locks will also block since write locks take priority.
As a result, if a query is long-running or timing out, all other queries
in the system will become blocked. Our default user request timeout is
90 seconds and some jobs have no timeout at all, so this is not a purely
theoretical concern, even if it's unlikely.

Refs: #2435
Co-authored-by: Jeff Mitchell <jeffrey.mitchell@gmail.com>

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

@@ -797,6 +797,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 {
@@ -843,6 +848,65 @@ func verifyKmsSetup(dbase *db.DB) 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 := acquireDatabaseSharedLock(c.Context, newDb)
+	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)
+	}
+
+	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
+}
+
 // acquireDatabaseSharedLock uses the schema manager to acquire a shared lock on
 // the database. This is done as a mechanism to disallow running migration commands
 // while the database is in use.

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/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
 				}(),