diff --git a/internal/cmd/commands/server/controller_db_swap_test.go b/internal/cmd/commands/server/controller_db_swap_test.go
new file mode 100644
index 000000000..580d5702a
--- /dev/null
+++ b/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
+}
diff --git a/internal/cmd/commands/server/listener_reload_test.go b/internal/cmd/commands/server/listener_reload_test.go
index 8d77baf28..fc56c97b4 100644
--- a/internal/cmd/commands/server/listener_reload_test.go
+++ b/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 {
diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go
index 897333149..b90f4272a 100644
--- a/internal/cmd/commands/server/server.go
+++ b/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.
diff --git a/internal/db/changesafe_reader_writer.go b/internal/db/changesafe_reader_writer.go
new file mode 100644
index 000000000..36fd0ead6
--- /dev/null
+++ b/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...)
+}
diff --git a/internal/db/db.go b/internal/db/db.go
index 279e488fd..c3f1546b1 100644
--- a/internal/db/db.go
+++ b/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
 }
diff --git a/internal/db/db_test.go b/internal/db/db_test.go
index 4db1bf0fc..a8020fa61 100644
--- a/internal/db/db_test.go
+++ b/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.
+		})
+	}
+}
diff --git a/internal/db/read_writer.go b/internal/db/read_writer.go
index cc3d2bde1..3035f988c 100644
--- a/internal/db/read_writer.go
+++ b/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
diff --git a/internal/db/read_writer_oplog.go b/internal/db/read_writer_oplog.go
index 630d3d6f3..75d95a32f 100644
--- a/internal/db/read_writer_oplog.go
+++ b/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...,
 	)
diff --git a/internal/db/testing.go b/internal/db/testing.go
index 54c5fe178..a0b29dd8f 100644
--- a/internal/db/testing.go
+++ b/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)
 }
 
diff --git a/internal/kms/kms.go b/internal/kms/kms.go
index c45082023..f2e3d28c6 100644
--- a/internal/kms/kms.go
+++ b/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() {
diff --git a/internal/kms/kms_ext_test.go b/internal/kms/kms_ext_test.go
index 3fe83598e..388cec58b 100644
--- a/internal/kms/kms_ext_test.go
+++ b/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 {
diff --git a/internal/kms/kms_test.go b/internal/kms/kms_test.go
index 0d6eb93db..7d61ca4d2 100644
--- a/internal/kms/kms_test.go
+++ b/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
 				}(),