Vet/race fixes (#4421)

This addresses all found vet/race fixes in community edition.

---------

Co-authored-by: Damian Debkowski <damian.debkowski@hashicorp.com>

.github/workflows/test-race.yml

@@ -0,0 +1,180 @@
+name: test-race
+
+on:
+  push:
+    paths-ignore:
+      - 'website/**'
+  workflow_call:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  DOCKER_MIRROR: docker.mirror.hashicorp.services
+
+jobs:
+  setup:
+    outputs:
+      go-version: ${{ steps.get-go-version.outputs.go-version }}
+      cache-go-build: ${{ steps.go-cache-paths.outputs.go-build }}
+      cache-go-mod: ${{ steps.go-cache-paths.outputs.go-mod }}
+      cache-go-bin: ${{ steps.go-cache-paths.outputs.go-bin }}
+      go-cache-key: ${{ steps.go-cache-key.outputs.key }}
+      plugin-cache-path: ${{ steps.plugin-cache-paths.outputs.path }}
+      plugin-cache-key: ${{ steps.plugin-cache-key.outputs.key }}
+    runs-on: ${{ fromJSON(vars.RUNNER) }}
+    steps:
+      - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
+        with:
+          fetch-depth: '0'
+      - name: Determine Go version
+        id: get-go-version
+        # We use .go-version as our source of truth for current Go
+        # version, because "goenv" can react to it automatically.
+        run: |
+          echo "Building with Go $(cat .go-version)"
+          echo "go-version=$(cat .go-version)" >> "$GITHUB_OUTPUT"
+      - name: Set up Go
+        uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0
+        with:
+          go-version: "${{ steps.get-go-version.outputs.go-version }}"
+          cache: false
+      - name: Determine go cache key
+        id: go-cache-key
+        run: |
+          echo "key=${{ runner.os }}-go-${{ hashFiles('**/go.sum', './Makefile', './tools/tools.go') }}" >> "$GITHUB_OUTPUT"
+      - name: Determine Go cache paths
+        id: go-cache-paths
+        run: |
+          echo "go-build=$(go env GOCACHE)" >> "$GITHUB_OUTPUT"
+          echo "go-mod=$(go env GOMODCACHE)" >> "$GITHUB_OUTPUT"
+          echo "go-bin=$(go env GOPATH)/bin" >> "$GITHUB_OUTPUT"
+      - name: Set up Go modules cache
+        uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0
+        with:
+          path: |
+            ${{ steps.go-cache-paths.outputs.go-build }}
+            ${{ steps.go-cache-paths.outputs.go-mod }}
+            ${{ steps.go-cache-paths.outputs.go-bin }}
+          key: ${{ steps.go-cache-key.outputs.key }}
+          restore-keys: |
+            ${{ runner.os }}-go
+      - name: Install Tools
+        run: |
+          go mod download
+          make tools
+      - name: Determine plugin cache key
+        id: plugin-cache-key
+        run: |
+          echo "key=${{ runner.os }}-plugins-${{ hashFiles('plugins/**/*.go', 'plugins/**/go.sum', './Makefile', './scripts/plugins.sh') }}" >> "$GITHUB_OUTPUT"
+      - name: Determin plugin cache path
+        id: plugin-cache-paths
+        run: |
+          echo "path=plugins/**/assets/*.gz" >> "$GITHUB_OUTPUT"
+      - name: Set up plugin cache
+        id: plugin-cache
+        uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0
+        with:
+          path: |
+            ${{ steps.plugin-cache-paths.outputs.path }}
+          key: ${{ steps.plugin-cache-key.outputs.key }}
+          restore-keys: |
+            ${{ runner.os }}-plugin
+      - name: Build Plugins
+        if: steps.plugin-cache.outputs.cache-hit != 'true'
+        run: |
+          make build-plugins
+  test-modules:
+    needs:
+      - setup
+    runs-on: ${{ fromJSON(vars.RUNNER) }}
+    strategy:
+      matrix:
+        module: ["api", "sdk"]
+    steps:
+      - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
+      - name: Set up go
+        uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0
+        with:
+          go-version: "${{ needs.setup.outputs.go-version }}"
+          cache: false
+      - name: Set up Go modules cache
+        uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0
+        with:
+          path: |
+            ${{ needs.setup.outputs.cache-go-build }}
+            ${{ needs.setup.outputs.cache-go-mod }}
+            ${{ needs.setup.outputs.cache-go-bin }}
+          key: ${{ needs.setup.outputs.go-cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-go
+          fail-on-cache-miss: true
+      - name: Test ${{ matrix.module }} Module
+        run: |
+          make test-${{ matrix.module }}
+
+  test:
+    needs:
+      - setup
+    runs-on: ${{ fromJSON(vars.RUNNER_LARGE) }}
+    steps:
+      - name: ulimit
+        run: |
+          echo "Soft limits"
+          ulimit -Sa
+          echo "Hard limits"
+          ulimit -Ha
+      - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
+      - name: Set up go
+        uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0
+        with:
+          go-version: "${{ needs.setup.outputs.go-version }}"
+          cache: false
+      - name: Set up Go modules cache
+        uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0
+        with:
+          path: |
+            ${{ needs.setup.outputs.cache-go-build }}
+            ${{ needs.setup.outputs.cache-go-mod }}
+            ${{ needs.setup.outputs.cache-go-bin }}
+          key: ${{ needs.setup.outputs.go-cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-go
+          fail-on-cache-miss: true
+      - name: Set up plugin cache
+        id: plugin-cache
+        uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0
+        with:
+          path: |
+            ${{ needs.setup.outputs.plugin-cache-path }}
+          key: ${{ needs.setup.outputs.plugin-cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-plugin
+      - name: GH fix for localhost resolution
+        if: github.repository == 'hashicorp/boundary'
+        run: |
+          cat /etc/hosts && echo "-----------"
+          sudo sed -i 's/::1 *localhost ip6-localhost ip6-loopback/::1 ip6 -localhost ip6-loopback/g' /etc/hosts
+          cat /etc/hosts
+      - name: Initialize Test Database
+        run: |
+          which pg_isready || sudo apt-get update && sudo apt-get install -y postgresql-client
+          make DOCKER_ARGS='-d' PG_OPTS='-c shared_buffers=256MB -c max_connections=200000' -C testing/dbtest/docker database-up
+          until pg_isready -h 127.0.0.1; do docker container inspect boundary-sql-tests &> /dev/null || exit 255; sleep 1; done
+
+      - name: Test
+        uses: nick-fields/retry@14672906e672a08bd6eeb15720e9ed3ce869cdd4 # v2.9.0
+        env:
+          TEST_PACKAGE: "./..."
+          TEST_TIMEOUT: 120m
+          TESTARGS: -race -v
+        with:
+          max_attempts: 1
+          timeout_minutes: 120
+          retry_on: error
+          command: make test
+      - name: Cleanup
+        if: always()
+        run: |
+          make -C testing/dbtest/docker clean

go.mod

@@ -19,8 +19,8 @@ require (
 	github.com/hashicorp/boundary/api v0.0.45
 	github.com/hashicorp/boundary/sdk v0.0.42
 	github.com/hashicorp/cap v0.5.0
-	github.com/hashicorp/dawdle v0.4.0
-	github.com/hashicorp/eventlogger v0.2.8
+	github.com/hashicorp/dawdle v0.5.0
+	github.com/hashicorp/eventlogger v0.2.9
 	github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231208142215-efdb51ec090d
 	github.com/hashicorp/go-bexpr v0.1.13
 	github.com/hashicorp/go-cleanhttp v0.5.2

go.sum

@@ -194,15 +194,15 @@ github.com/hashicorp/cap v0.5.0 h1:YIlAYxdXXtx2IL1JDvP2OyEl55Ooi0yl573kSB9Orlw=
 github.com/hashicorp/cap v0.5.0/go.mod h1:IAy00Er+ZFpMo+5x6B4bkO2HgpzgrkfsuDWMmHAuKUE=
 github.com/hashicorp/cap/ldap v0.0.0-20240206183135-ed8f24513744 h1:6cU01ORaCUxRXRS0YYnmm4tMfAvFRjamhjgMrgmoWZs=
 github.com/hashicorp/cap/ldap v0.0.0-20240206183135-ed8f24513744/go.mod h1:SOJuo/OFD0qKQ7d8qOy+a2B9Tw0/97oHIo+qXxs7HZE=
-github.com/hashicorp/dawdle v0.4.0 h1:nAE+aCoAmVulgCPPPrp6oGOykjPhxEe4VUNms/mTKeA=
-github.com/hashicorp/dawdle v0.4.0/go.mod h1:gOBY8rFXpW4Rw39up1FIacw/1pG/fO779LfchCY3KUw=
+github.com/hashicorp/dawdle v0.5.0 h1:cPKzDUfuq5JhhAcChg8cygWstV+WQ7MSn7FSP7AtPnI=
+github.com/hashicorp/dawdle v0.5.0/go.mod h1:gOBY8rFXpW4Rw39up1FIacw/1pG/fO779LfchCY3KUw=
 github.com/hashicorp/dbassert v0.0.0-20231012105025-1bc1bd88e22b h1:rvg3eCucdlpfsnbYXM1Ett4f0reNmIuCruyJUF4iw/0=
 github.com/hashicorp/dbassert v0.0.0-20231012105025-1bc1bd88e22b/go.mod h1:GCYXHZsaYdDfPh/xD3A/t6754HPpumeLkA0vLpYVjZE=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
-github.com/hashicorp/eventlogger v0.2.8 h1:WITwN1QGMtxHQE7l69WBwLJIJA5vH5pcvX69peURU14=
-github.com/hashicorp/eventlogger v0.2.8/go.mod h1://CHt6/j+Q2lc0NlUB5af4aS2M0c0aVBg9/JfcpAyhM=
+github.com/hashicorp/eventlogger v0.2.9 h1:QO8hPXNngadMp72FVNTwIduLAG9fcLP7t59bSFd7gDY=
+github.com/hashicorp/eventlogger v0.2.9/go.mod h1://CHt6/j+Q2lc0NlUB5af4aS2M0c0aVBg9/JfcpAyhM=
 github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231208142215-efdb51ec090d h1:Zia/aoFQmhtzqY6NVR0DVcxmZowbavqbTeum1bRRZhs=
 github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231208142215-efdb51ec090d/go.mod h1:tMywUTIvdB/FXhwm6HMTt61C8/eODY6gitCHhXtyojg=
 github.com/hashicorp/go-bexpr v0.1.13 h1:HNwp7vZrMpRq8VZXj8VF90LbZpRjQQpim1oJF0DgSwg=

internal/auth/ldap/service_list_accounts_ext_test.go

@@ -260,7 +260,7 @@ func TestService_ListAccounts(t *testing.T) {
 			filterFunc := func(_ context.Context, a auth.Account) (bool, error) {
 				return true, nil
 			}
-			_, err = ldap.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := ldap.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("nil repo", func(t *testing.T) {

internal/auth/ldap/service_list_managed_groups_ext_test.go

@@ -175,7 +175,7 @@ func TestService_ListManagedGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) {
 				return true, nil
 			}
-			_, err = ldap.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := ldap.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -263,7 +263,7 @@ func TestService_ListManagedGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) {
 				return true, nil
 			}
-			_, err = ldap.ListManagedGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := ldap.ListManagedGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("nil repo", func(t *testing.T) {
@@ -341,7 +341,7 @@ func TestService_ListManagedGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) {
 				return true, nil
 			}
-			_, err = ldap.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := ldap.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/auth/oidc/service_list_accounts_ext_test.go

@@ -266,7 +266,7 @@ func TestService_ListAccounts(t *testing.T) {
 			filterFunc := func(_ context.Context, a auth.Account) (bool, error) {
 				return true, nil
 			}
-			_, err = oidc.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := oidc.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("nil repo", func(t *testing.T) {

internal/auth/oidc/service_list_managed_groups_ext_test.go

@@ -181,7 +181,7 @@ func TestService_ListManagedGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) {
 				return true, nil
 			}
-			_, err = oidc.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := oidc.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -347,7 +347,7 @@ func TestService_ListManagedGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) {
 				return true, nil
 			}
-			_, err = oidc.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := oidc.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/auth/password/service_list_accounts_ext_test.go

@@ -162,7 +162,7 @@ func TestService_ListAccounts(t *testing.T) {
 			filterFunc := func(_ context.Context, a auth.Account) (bool, error) {
 				return true, nil
 			}
-			_, err = password.ListAccountsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
+			_, err := password.ListAccountsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/authtoken/service_list_ext_test.go

@@ -155,7 +155,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, at *authtoken.AuthToken) (bool, error) {
 				return true, nil
 			}
-			_, err = authtoken.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()})
+			_, err := authtoken.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()})
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -243,7 +243,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, at *authtoken.AuthToken) (bool, error) {
 				return true, nil
 			}
-			_, err = authtoken.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()})
+			_, err := authtoken.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()})
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/bsr/testing.go

@@ -11,11 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-var registerTestProtocol *sync.Once
-
-func init() {
-	registerTestProtocol = new(sync.Once)
-}
+var registerTestProtocol *sync.Once = new(sync.Once)
 
 // TestRegisterSummaryAllocFunc registers "TEST" as a protocol for all container types
 // The channel summary will include the following:

internal/clientcache/internal/daemon/server.go

@@ -14,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/boundary/api"
@@ -62,7 +63,7 @@ type CacheServer struct {
 	info     map[string]string
 
 	storeUrl string
-	store    *db.DB
+	store    atomic.Pointer[db.DB]
 
 	tickerWg *sync.WaitGroup
 	httpSrv  *http.Server
@@ -224,9 +225,13 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e
 	s.info["Store debug"] = strconv.FormatBool(s.conf.StoreDebug)
 	s.infoKeys = append(s.infoKeys, "Store debug")
 
-	if s.store, s.storeUrl, err = openStore(ctx, s.conf.DatabaseUrl, s.conf.StoreDebug); err != nil {
+	var store *db.DB
+	store, s.storeUrl, err = openStore(ctx, s.conf.DatabaseUrl, s.conf.StoreDebug)
+	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
+	s.store.Store(store)
+
 	if s.storeUrl != "" {
 		s.info["Database URL"] = s.storeUrl
 		s.infoKeys = append(s.infoKeys, "Database URL")
@@ -260,7 +265,7 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e
 
 	s.printInfo(ctx)
 
-	repo, err := cache.NewRepository(ctx, s.store, &sync.Map{}, cmd.ReadTokenFromKeyring, opts.withBoundaryTokenReaderFunc)
+	repo, err := cache.NewRepository(ctx, s.store.Load(), &sync.Map{}, cmd.ReadTokenFromKeyring, opts.withBoundaryTokenReaderFunc)
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}

internal/clientcache/internal/daemon/testing.go

@@ -71,7 +71,7 @@ func (s *TestServer) Serve(t *testing.T, opt ...Option) error {
 func (s *TestServer) AddKeyringToken(t *testing.T, address, keyring, tokenName, tokenId string, atReadFn cache.BoundaryTokenReaderFn) {
 	t.Helper()
 	ctx := context.Background()
-	r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn)
+	r, err := cache.NewRepository(ctx, s.CacheServer.store.Load(), &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn)
 	require.NoError(t, err)
 
 	require.NoError(t, r.AddKeyringToken(ctx, address, cache.KeyringToken{
@@ -86,7 +86,7 @@ func (s *TestServer) AddKeyringToken(t *testing.T, address, keyring, tokenName,
 func (s *TestServer) AddResources(t *testing.T, p *authtokens.AuthToken, tars []*targets.Target, sess []*sessions.Session, atReadFn cache.BoundaryTokenReaderFn) {
 	t.Helper()
 	ctx := context.Background()
-	r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn)
+	r, err := cache.NewRepository(ctx, s.CacheServer.store.Load(), &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn)
 	require.NoError(t, err)
 
 	tarFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*targets.Target, []string, cache.RefreshTokenValue, error) {
@@ -113,7 +113,7 @@ func (s *TestServer) AddResources(t *testing.T, p *authtokens.AuthToken, tars []
 func (s *TestServer) AddUnsupportedCachingData(t *testing.T, p *authtokens.AuthToken, atReadFn cache.BoundaryTokenReaderFn) {
 	t.Helper()
 	ctx := context.Background()
-	r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn)
+	r, err := cache.NewRepository(ctx, s.CacheServer.store.Load(), &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn)
 	require.NoError(t, err)
 
 	tarFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*targets.Target, []string, cache.RefreshTokenValue, error) {

internal/clientcache/internal/daemon/ticker.go

@@ -37,7 +37,6 @@ type refreshTicker struct {
 	refreshInterval        time.Duration
 	recheckSupportInterval time.Duration
 	randomizationFactor    float64
-	rand                   *rand.Rand
 	refreshChan            chan struct{}
 	refresher              refreshService
 }
@@ -80,7 +79,6 @@ func newRefreshTicker(ctx context.Context, refresh refreshService, opt ...Option
 		refreshInterval:        refreshInterval,
 		recheckSupportInterval: recheckSupportInterval,
 		randomizationFactor:    randomizationFactor,
-		rand:                   rand.New(rand.NewSource(time.Now().UnixNano())),
 		refresher:              refresh,
 
 		// We make this channel size 1 so if something happens midway through the refresh
@@ -151,12 +149,12 @@ func (rt *refreshTicker) nextIntervalWithRandomness(d time.Duration) time.Durati
 		return d
 	}
 
-	randFactor := rt.rand.Float64()
+	randFactor := rand.Float64()
 	randFactor = randFactor * rt.randomizationFactor
 
 	randDur := randFactor * float64(d)
 	// Half a chance to be faster, not slower
-	if rt.rand.Float32() > 0.5 {
+	if rand.Float32() > 0.5 {
 		randDur = -1 * randDur
 	}
 	return d + time.Duration(randDur)

internal/cmd/base/logging.go

@@ -7,9 +7,11 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync/atomic"
 
 	"github.com/hashicorp/boundary/internal/cmd/base/logging"
 	"github.com/hashicorp/go-hclog"
+	"google.golang.org/grpc/grpclog"
 )
 
 func ProcessLogLevelAndFormat(flagLogLevel, flagLogFormat, configLogLevel, configLogFormat string) (hclog.Level, logging.LogFormat, error) {
@@ -59,77 +61,140 @@ func ProcessLogLevelAndFormat(flagLogLevel, flagLogFormat, configLogLevel, confi
 	return level, logFormat, nil
 }
 
+type grpcLogFakerLogHolder struct {
+	logger hclog.Logger
+}
+
+var grpcLogFaker *GRPCLogFaker = &GRPCLogFaker{
+	loggingEnabled: new(atomic.Bool),
+	logger:         new(atomic.Pointer[grpcLogFakerLogHolder]),
+}
+
+func init() {
+	grpclog.SetLoggerV2(grpcLogFaker)
+}
+
 type GRPCLogFaker struct {
-	Logger hclog.Logger
-	Log    bool
+	logger         *atomic.Pointer[grpcLogFakerLogHolder]
+	loggingEnabled *atomic.Bool
+}
+
+func (g *GRPCLogFaker) SetLogOnOff(on bool) {
+	g.loggingEnabled.Store(on)
+}
+
+func (g *GRPCLogFaker) SetLogger(logger hclog.Logger) {
+	g.logger.Store(&grpcLogFakerLogHolder{logger: logger})
 }
 
 func (g *GRPCLogFaker) Fatal(args ...any) {
-	g.Logger.Error(fmt.Sprint(args...))
+	if l := g.logger.Load(); l != nil {
+		l.logger.Error(fmt.Sprint(args...))
+	}
 	os.Exit(1)
 }
 
 func (g *GRPCLogFaker) Fatalf(format string, args ...any) {
-	g.Logger.Error(fmt.Sprintf(format, args...))
+	if l := g.logger.Load(); l != nil {
+		l.logger.Error(fmt.Sprintf(format, args...))
+	}
 	os.Exit(1)
 }
 
 func (g *GRPCLogFaker) Fatalln(args ...any) {
-	g.Logger.Error(fmt.Sprintln(args...))
+	if l := g.logger.Load(); l != nil {
+		l.logger.Error(fmt.Sprintln(args...))
+	}
 	os.Exit(1)
 }
 
 func (g *GRPCLogFaker) Info(args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprint(args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprint(args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Infof(format string, args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprintf(format, args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprintf(format, args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Infoln(args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprintln(args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprintln(args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Warning(args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprint(args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprint(args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Warningf(format string, args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprintf(format, args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprintf(format, args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Warningln(args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprintln(args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprintln(args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Error(args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprint(args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprint(args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Errorf(format string, args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprintf(format, args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprintf(format, args...))
+			}
+		}
 	}
 }
 
 func (g *GRPCLogFaker) Errorln(args ...any) {
-	if g.Log && g.Logger.IsDebug() {
-		g.Logger.Debug(fmt.Sprintln(args...))
+	if g.loggingEnabled.Load() {
+		if l := g.logger.Load(); l != nil {
+			if l.logger.IsDebug() {
+				l.logger.Debug(fmt.Sprintln(args...))
+			}
+		}
 	}
 }
 

internal/cmd/base/servers.go

@@ -50,7 +50,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/text/cases"
 	"golang.org/x/text/language"
-	"google.golang.org/grpc/grpclog"
 )
 
 const (
@@ -285,10 +284,8 @@ func (b *Server) SetupLogging(flagLogLevel, flagLogFormat, configLogLevel, confi
 
 	// create GRPC logger
 	namedGRPCLogFaker := b.Logger.Named("grpclogfaker")
-	grpclog.SetLoggerV2(&GRPCLogFaker{
-		Logger: namedGRPCLogFaker,
-		Log:    os.Getenv("BOUNDARY_GRPC_LOGGING") != "",
-	})
+	grpcLogFaker.SetLogOnOff(os.Getenv("BOUNDARY_GRPC_LOGGING") != "")
+	grpcLogFaker.SetLogger(namedGRPCLogFaker)
 
 	b.Info["log level"] = logLevel.String()
 	b.InfoKeys = append(b.InfoKeys, "log level")

internal/credential/service_list_credentials_ext_test.go

@@ -182,7 +182,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, c credential.Static) (bool, error) {
 				return true, nil
 			}
-			_, err = credential.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId())
+			_, err := credential.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -270,7 +270,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, c credential.Static) (bool, error) {
 				return true, nil
 			}
-			_, err = credential.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId())
+			_, err := credential.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("nil repo", func(t *testing.T) {

internal/credential/service_list_libraries_ext_test.go

@@ -331,7 +331,7 @@ func TestLibraryService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, l credential.Library) (bool, error) {
 				return true, nil
 			}
-			_, err = credential.ListLibrariesRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId())
+			_, err := credential.ListLibrariesRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/credential/service_list_stores_ext_test.go

@@ -370,7 +370,7 @@ func TestStoreService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, s credential.Store) (bool, error) {
 				return true, nil
 			}
-			_, err = credential.ListStoresRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId})
+			_, err := credential.ListStoresRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId})
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/daemon/cluster/handlers/testing.go

@@ -87,12 +87,12 @@ func TestRegisterHandlerFn(t *testing.T, msgType pbs.MsgType, h UpstreamMessageH
 	return func(t *testing.T) {
 		t.Helper()
 		testCtx := context.Background()
-		var cpHandlerRegistry sync.Map
+		cpHandlerRegistry := new(sync.Map)
 		upstreamMessageHandler.Range(func(k, v interface{}) bool {
 			cpHandlerRegistry.Store(k, v)
 			return true
 		})
-		var cpTypeSpecifierRegister sync.Map
+		cpTypeSpecifierRegister := new(sync.Map)
 		upstreamMessageTypeSpecifier.Range(func(k, v interface{}) bool {
 			cpTypeSpecifierRegister.Store(k, v)
 			return true

internal/daemon/cluster/handlers/upstream_message_service_controller.go

@@ -19,7 +19,7 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-var upstreamMessageHandler sync.Map
+var upstreamMessageHandler *sync.Map = new(sync.Map)
 
 // UpstreamMessageHandler defines a handler for an UpstreamMessageRequest(s).
 //

internal/daemon/cluster/handlers/upstream_message_service_test.go

@@ -65,7 +65,7 @@ func Test_RegisterUpstreamMessageHandler(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
-			var cp sync.Map
+			cp := new(sync.Map)
 			upstreamMessageHandler.Range(func(k, v interface{}) bool {
 				cp.Store(k, v)
 				return true
@@ -468,7 +468,7 @@ func Test_RegisterUpstreamMessageTypeSpecifier(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
-			var cp sync.Map
+			cp := new(sync.Map)
 			upstreamMessageTypeSpecifier.Range(func(k, v interface{}) bool {
 				cp.Store(k, v)
 				return true

internal/daemon/cluster/handlers/upstream_message_service_worker.go

@@ -82,7 +82,7 @@ type UpstreamMessageTypeSpecifier interface {
 	AllocResponse() proto.Message
 }
 
-var upstreamMessageTypeSpecifier sync.Map
+var upstreamMessageTypeSpecifier *sync.Map = new(sync.Map)
 
 // registerUpstreamMessageTypeSpecifier will register an
 // UpstreamMessageTypeSpecifier for the specified msg name.

internal/daemon/worker/auth_rotation_test.go

@@ -170,8 +170,8 @@ func TestRotationTicking(t *testing.T) {
 		// Stop and start the client connections to ensure the new credentials
 		// are valid; if not, we won't establish a new connection and rotation
 		// will fail
-		require.NotNil(w.Worker().GrpcClientConn)
-		require.NoError(w.Worker().GrpcClientConn.Close())
+		require.NotNil(w.Worker().GrpcClientConn.Load())
+		require.NoError(w.Worker().GrpcClientConn.Load().Close())
 		require.NoError(w.Worker().StartControllerConnections())
 		rotationCount++
 	}

internal/daemon/worker/controller_connection.go

@@ -189,7 +189,7 @@ func (w *Worker) createClientConn(addr string) error {
 		return fmt.Errorf("error dialing controller for worker auth: %w", err)
 	}
 
-	w.GrpcClientConn = cc
+	w.GrpcClientConn.Store(cc)
 	w.controllerMultihopConn.Store(multihop.NewMultihopServiceClient(cc))
 
 	var producer handlers.UpstreamMessageServiceClientProducer

internal/daemon/worker/controller_connection_test.go

@@ -6,6 +6,7 @@ package worker
 import (
 	"context"
 	"fmt"
+	"net"
 	"strconv"
 	"sync/atomic"
 	"testing"
@@ -119,8 +120,9 @@ func waitForConnectionStateCondition(upstreamConnectionState *atomic.Value, expe
 }
 
 type serverTestInfo struct {
-	srv     *grpc.Server
-	address string
+	srv      *grpc.Server
+	address  string
+	listener net.Listener
 }
 
 func createTestServers(t *testing.T) ([]*serverTestInfo, error) {
@@ -133,10 +135,12 @@ func createTestServers(t *testing.T) ([]*serverTestInfo, error) {
 			return nil, err
 		}
 		srv := grpc.NewServer()
-		lInfo := &serverTestInfo{srv: srv, address: listener.Addr().String()}
+		lInfo := &serverTestInfo{srv: srv, listener: listener, address: listener.Addr().String()}
 		servers = append(servers, lInfo)
-		go func(i int) {
-			if err := servers[i].srv.Serve(listener); err != nil {
+	}
+	for i := range servers {
+		go func(num int) {
+			if err := servers[num].srv.Serve(servers[num].listener); err != nil {
 				t.Logf("error serving: %v", err)
 			}
 		}(i)

internal/daemon/worker/countingconn_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -20,54 +21,29 @@ func TestCountingConn(t *testing.T) {
 		underlyingConn *testNetConn
 	}{
 		{
-			name:       "noErrors",
-			writeBytes: []byte("hello"),
-			underlyingConn: &testNetConn{
-				bytesToRead: 100,
-				readErr:     false,
-				writeErr:    false,
-				closeErr:    false,
-			},
+			name:           "noErrors",
+			writeBytes:     []byte("hello"),
+			underlyingConn: newTestNetConn(100, false, false, false),
 		},
 		{
-			name:       "readErr",
-			writeBytes: []byte("hello"),
-			underlyingConn: &testNetConn{
-				bytesToRead: 100,
-				readErr:     true,
-				writeErr:    false,
-				closeErr:    false,
-			},
+			name:           "readErr",
+			writeBytes:     []byte("hello"),
+			underlyingConn: newTestNetConn(100, true, false, false),
 		},
 		{
-			name:       "writeErr",
-			writeBytes: []byte("hello"),
-			underlyingConn: &testNetConn{
-				bytesToRead: 100,
-				readErr:     false,
-				writeErr:    true,
-				closeErr:    false,
-			},
+			name:           "writeErr",
+			writeBytes:     []byte("hello"),
+			underlyingConn: newTestNetConn(100, false, true, false),
 		},
 		{
-			name:       "closeErr",
-			writeBytes: []byte("hello"),
-			underlyingConn: &testNetConn{
-				bytesToRead: 100,
-				readErr:     false,
-				writeErr:    false,
-				closeErr:    true,
-			},
+			name:           "closeErr",
+			writeBytes:     []byte("hello"),
+			underlyingConn: newTestNetConn(100, false, false, true),
 		},
 		{
-			name:       "allErr",
-			writeBytes: []byte("hello"),
-			underlyingConn: &testNetConn{
-				bytesToRead: 100,
-				readErr:     true,
-				writeErr:    true,
-				closeErr:    true,
-			},
+			name:           "allErr",
+			writeBytes:     []byte("hello"),
+			underlyingConn: newTestNetConn(100, true, true, true),
 		},
 	}
 
@@ -77,7 +53,7 @@ func TestCountingConn(t *testing.T) {
 
 			readBytes := make([]byte, tt.underlyingConn.bytesToRead)
 			read, err := conn.Read(readBytes)
-			require.True(t, tt.underlyingConn.readCalled)
+			require.True(t, tt.underlyingConn.readCalled.Load())
 			if tt.underlyingConn.readErr {
 				require.Error(t, err)
 				require.Equal(t, 1, read)
@@ -90,7 +66,7 @@ func TestCountingConn(t *testing.T) {
 			}
 
 			written, err := conn.Write(tt.writeBytes)
-			require.True(t, tt.underlyingConn.writeCalled)
+			require.True(t, tt.underlyingConn.writeCalled.Load())
 			if tt.underlyingConn.writeErr {
 				require.Error(t, err)
 				require.Equal(t, 1, written)
@@ -102,7 +78,7 @@ func TestCountingConn(t *testing.T) {
 			}
 
 			err = conn.Close()
-			require.True(t, tt.underlyingConn.closeCalled)
+			require.True(t, tt.underlyingConn.closeCalled.Load())
 			if tt.underlyingConn.closeErr {
 				require.Error(t, err)
 			} else {
@@ -121,7 +97,7 @@ func TestCountingConnConcurrentCalls(t *testing.T) {
 	concurrentReads := 1000
 	concurrentWrites := 1000
 
-	conn := &countingConn{Conn: &testNetConn{bytesToRead: bytesToRead}}
+	conn := &countingConn{Conn: newTestNetConn(bytesToRead, false, false, false)}
 
 	wg := sync.WaitGroup{}
 	wg.Add(concurrentReads)
@@ -164,13 +140,25 @@ type testNetConn struct {
 	closeErr    bool
 
 	// Test results
-	readCalled  bool
-	writeCalled bool
-	closeCalled bool
+	readCalled  *atomic.Bool
+	writeCalled *atomic.Bool
+	closeCalled *atomic.Bool
+}
+
+func newTestNetConn(bytesToRead int, readErr, writeErr, closeErr bool) *testNetConn {
+	return &testNetConn{
+		bytesToRead: bytesToRead,
+		readErr:     readErr,
+		writeErr:    writeErr,
+		closeErr:    closeErr,
+		readCalled:  new(atomic.Bool),
+		writeCalled: new(atomic.Bool),
+		closeCalled: new(atomic.Bool),
+	}
 }
 
 func (t *testNetConn) Read(in []byte) (int, error) {
-	t.readCalled = true
+	t.readCalled.Store(true)
 	if t.readErr {
 		return 1, fmt.Errorf("oops, read error")
 	}
@@ -183,7 +171,7 @@ func (t *testNetConn) Read(in []byte) (int, error) {
 }
 
 func (t *testNetConn) Write(in []byte) (int, error) {
-	t.writeCalled = true
+	t.writeCalled.Store(true)
 	if t.writeErr {
 		return 1, fmt.Errorf("oops, write error")
 	}
@@ -192,7 +180,7 @@ func (t *testNetConn) Write(in []byte) (int, error) {
 }
 
 func (t *testNetConn) Close() error {
-	t.closeCalled = true
+	t.closeCalled.Store(true)
 	if t.closeErr {
 		return fmt.Errorf("oops, close error")
 	}

internal/daemon/worker/listeners_test.go

@@ -89,7 +89,7 @@ func TestStartListeners(t *testing.T) {
 			}
 			w.controllerUpstreamMsgConn.Store(&dummyClientProducer)
 
-			manager, err := session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn))
+			manager, err := session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn.Load()))
 			require.NoError(t, err)
 			err = w.startListeners(manager)
 			if tt.expErr {

internal/daemon/worker/proxy/proxy.go

@@ -17,7 +17,7 @@ var (
 	TcpHandlerName = "tcp"
 
 	// handlers is the map of registered handlers
-	handlers sync.Map
+	handlers *sync.Map = new(sync.Map)
 
 	// ErrUnknownProtocol specifies the provided protocol has no registered handler
 	ErrUnknownProtocol = errors.New("proxy: handler not found for protocol")

internal/daemon/worker/proxy/proxy_conn_tracker.go

@@ -9,10 +9,10 @@ import (
 )
 
 // ProxyState contains the current state of proxies in this process.
-var ProxyState proxyState
+var ProxyState proxyState = proxyState{proxyCount: new(atomic.Int64)}
 
 type proxyState struct {
-	proxyCount atomic.Int64
+	proxyCount *atomic.Int64
 }
 
 // CurrentProxiedConnections returns the current number of ongoing proxied

internal/daemon/worker/proxy/proxy_conn_tracker_test.go

@@ -6,6 +6,7 @@ package proxy
 import (
 	"net/http"
 	"net/http/httptest"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -14,7 +15,7 @@ import (
 
 func TestProxyStateHelpers(t *testing.T) {
 	old := ProxyState
-	ProxyState = proxyState{}
+	ProxyState = proxyState{proxyCount: new(atomic.Int64)}
 	t.Cleanup(func() {
 		ProxyState = old
 	})

internal/daemon/worker/proxy/proxy_test.go

@@ -24,7 +24,7 @@ func TestRegisterHandler(t *testing.T) {
 	t.Cleanup(func() {
 		handlers = oldHandler
 	})
-	handlers = sync.Map{}
+	handlers = new(sync.Map)
 
 	err := RegisterHandler("protocol", fn)
 	require.NoError(err)
@@ -47,7 +47,7 @@ func TestAlwaysTcpGetHandler(t *testing.T) {
 	t.Cleanup(func() {
 		handlers = oldHandler
 	})
-	handlers = sync.Map{}
+	handlers = new(sync.Map)
 	_, err := tcpOnly("wid", nil)
 	assert.ErrorIs(err, ErrUnknownProtocol)
 

internal/daemon/worker/rpc_registration.go

@@ -37,7 +37,7 @@ func registerWorkerStatusSessionService(ctx context.Context, w *Worker, server *
 		return fmt.Errorf("%s: server is nil", op)
 	}
 
-	statusSessionService := NewWorkerProxyServiceServer(w.GrpcClientConn)
+	statusSessionService := NewWorkerProxyServiceServer(w.GrpcClientConn.Load())
 	pbs.RegisterServerCoordinationServiceServer(server, statusSessionService)
 	pbs.RegisterSessionServiceServer(server, statusSessionService)
 	return nil

internal/daemon/worker/session/manager.go

@@ -61,7 +61,7 @@ type Manager interface {
 
 type manager struct {
 	controllerSessionConn pbs.SessionServiceClient
-	sessionMap            sync.Map
+	sessionMap            *sync.Map
 }
 
 var _ Manager = (*manager)(nil)
@@ -74,6 +74,7 @@ func NewManager(client pbs.SessionServiceClient) (*manager, error) {
 	}
 	return &manager{
 		controllerSessionConn: client,
+		sessionMap:            new(sync.Map),
 	}, nil
 }
 

internal/daemon/worker/session/manager_test.go

@@ -273,7 +273,7 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 	cases := []struct {
 		name             string
 		sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData
-		sessionInfoMap   func() sync.Map
+		sessionInfoMap   func() *sync.Map
 		expected         []string
 		expectedClosed   map[string]struct{}
 		expectedErr      []error
@@ -288,8 +288,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 					{ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED},
 				},
 			},
-			sessionInfoMap: func() sync.Map {
-				var m sync.Map
+			sessionInfoMap: func() *sync.Map {
+				m := new(sync.Map)
 				m.Store("one", &sess{
 					resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{
 						SessionId: "one",
@@ -333,8 +333,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 					{ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED},
 				},
 			},
-			sessionInfoMap: func() sync.Map {
-				var m sync.Map
+			sessionInfoMap: func() *sync.Map {
+				m := new(sync.Map)
 				m.Store("one", &sess{
 					resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{
 						SessionId: "one",
@@ -369,8 +369,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 					{ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED},
 				},
 			},
-			sessionInfoMap: func() sync.Map {
-				var m sync.Map
+			sessionInfoMap: func() *sync.Map {
+				m := new(sync.Map)
 				m.Store("one", &sess{
 					resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{
 						SessionId: "one",
@@ -400,8 +400,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 					{ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED},
 				},
 			},
-			sessionInfoMap: func() sync.Map {
-				var m sync.Map
+			sessionInfoMap: func() *sync.Map {
+				m := new(sync.Map)
 				m.Store("one", &sess{
 					resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{
 						SessionId: "one",
@@ -431,8 +431,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 		{
 			name:             "empty",
 			sessionCloseInfo: make(map[string][]*pbs.CloseConnectionResponseData),
-			sessionInfoMap: func() sync.Map {
-				var m sync.Map
+			sessionInfoMap: func() *sync.Map {
+				m := new(sync.Map)
 				m.Store("one", &sess{
 					resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{
 						SessionId: "one",
@@ -453,7 +453,7 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			require := require.New(t)
-			manager := &manager{}
+			manager := &manager{sessionMap: new(sync.Map)}
 			manager.sessionMap = tc.sessionInfoMap()
 			actual, actualErr := setCloseTimeForResponse(manager, tc.sessionCloseInfo)
 

internal/daemon/worker/status.go

@@ -181,7 +181,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 	})
 
 	// Send status information
-	client := pbs.NewServerCoordinationServiceClient(w.GrpcClientConn)
+	client := pbs.NewServerCoordinationServiceClient(w.GrpcClientConn.Load())
 	var tags []*pb.TagPair
 	// If we're not going to request a tag update, no reason to have these
 	// marshaled on every status call.
@@ -242,7 +242,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 
 			// In the case that the control plane has gone down and come up with different IPs,
 			// append initial upstreams/ cluster addr to the resolver to try
-			if w.GrpcClientConn.GetState() == connectivity.TransientFailure {
+			if w.GrpcClientConn.Load().GetState() == connectivity.TransientFailure {
 				lastStatus := w.lastStatusSuccess.Load().(*LastStatusInformation)
 				if lastStatus != nil && lastStatus.LastCalculatedUpstreams != nil {
 					addrs := lastStatus.LastCalculatedUpstreams

internal/daemon/worker/worker.go

@@ -116,10 +116,11 @@ type Worker struct {
 
 	tickerWg sync.WaitGroup
 
-	// grpc.ClientConns are thread safe.
-	// See https://github.com/grpc/grpc-go/blob/master/Documentation/concurrency.md#clients
-	// This is exported for tests.
-	GrpcClientConn *grpc.ClientConn
+	// grpc.ClientConns are thread safe. See
+	// https://github.com/grpc/grpc-go/blob/master/Documentation/concurrency.md#clients
+	// However this is an atomic because we sometimes swap this pointer out
+	// (mostly in tests) - which isn't thread safe. This is exported for tests.
+	GrpcClientConn atomic.Pointer[grpc.ClientConn]
 
 	// receives address updates and contains the grpc resolver.
 	addressReceivers []addressReceiver
@@ -511,7 +512,7 @@ func (w *Worker) Start() error {
 		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error making controller connections"))
 	}
 
-	w.sessionManager, err = session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn))
+	w.sessionManager, err = session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn.Load()))
 	if err != nil {
 		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error creating session manager"))
 	}

internal/daemon/worker/workerdisconnect_test.go

@@ -82,7 +82,7 @@ func TestDeleteConnectedWorkers(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			prevState := tc.testWorker.Worker().GrpcClientConn.GetState()
+			prevState := tc.testWorker.Worker().GrpcClientConn.Load().GetState()
 			require.NotEqual(t, connectivity.TransientFailure, prevState)
 			require.NotEqual(t, connectivity.Shutdown, prevState)
 			_, err = serverRepo.DeleteWorker(ctx, tc.workerId)
@@ -90,12 +90,12 @@ func TestDeleteConnectedWorkers(t *testing.T) {
 			stateChangeCtx, cancel := context.WithTimeout(ctx, 4*time.Second)
 			defer cancel()
 			for {
-				tc.testWorker.Worker().GrpcClientConn.ResetConnectBackoff()
-				if !tc.testWorker.Worker().GrpcClientConn.WaitForStateChange(stateChangeCtx, prevState) {
+				tc.testWorker.Worker().GrpcClientConn.Load().ResetConnectBackoff()
+				if !tc.testWorker.Worker().GrpcClientConn.Load().WaitForStateChange(stateChangeCtx, prevState) {
 					assert.Fail(t, "State didn't change before context timed out")
 					break
 				}
-				newState := tc.testWorker.Worker().GrpcClientConn.GetState()
+				newState := tc.testWorker.Worker().GrpcClientConn.Load().GetState()
 				t.Logf("Changed from previous state: %s to new state: %s", prevState, newState)
 				if newState == connectivity.Shutdown || newState == connectivity.TransientFailure {
 					break

internal/event/eventer.go

@@ -118,7 +118,8 @@ var (
 )
 
 func defaultLogger() hclog.Logger {
-	opts := hclog.DefaultOptions
+	opts := new(hclog.LoggerOptions)
+	*opts = *hclog.DefaultOptions
 	opts.JSONFormat = true
 	return hclog.New(opts)
 }

internal/host/plugin/service_list_host_sets_ext_test.go

@@ -169,7 +169,7 @@ func TestService_ListHostSets(t *testing.T) {
 			filterFunc := func(_ context.Context, s host.Set, plg *plugin.Plugin) (bool, error) {
 				return true, nil
 			}
-			_, _, err = hostplugin.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId())
+			_, _, err := hostplugin.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId())
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/host/service_list_catalogs_ext_test.go

@@ -209,7 +209,7 @@ func TestCatalogService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, c host.Catalog, plgs map[string]*plugin.Plugin) (bool, error) {
 				return true, nil
 			}
-			_, _, err = host.ListCatalogsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId})
+			_, _, err := host.ListCatalogsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId})
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -297,7 +297,7 @@ func TestCatalogService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, c host.Catalog, plgs map[string]*plugin.Plugin) (bool, error) {
 				return true, nil
 			}
-			_, _, err = host.ListCatalogsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId})
+			_, _, err := host.ListCatalogsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId})
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("missing repo", func(t *testing.T) {

internal/iam/service_list_ext_test.go

@@ -839,7 +839,7 @@ func TestService_ListUsers(t *testing.T) {
 			filterFunc := func(_ context.Context, r *iam.User) (bool, error) {
 				return true, nil
 			}
-			_, err = iam.ListUsersRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()})
+			_, err := iam.ListUsersRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()})
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -1368,7 +1368,7 @@ func TestService_ListGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, r *iam.Group) (bool, error) {
 				return true, nil
 			}
-			_, err = iam.ListGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes)
+			_, err := iam.ListGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes)
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -1456,7 +1456,7 @@ func TestService_ListGroups(t *testing.T) {
 			filterFunc := func(_ context.Context, r *iam.Group) (bool, error) {
 				return true, nil
 			}
-			_, err = iam.ListGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes)
+			_, err := iam.ListGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes)
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/perms/grants.go

@@ -178,7 +178,7 @@ func (g Grant) CanonicalString() string {
 }
 
 // MarshalJSON provides a custom marshaller for grants
-func (g Grant) MarshalJSON(ctx context.Context) ([]byte, error) {
+func (g Grant) MarshalJSON() ([]byte, error) {
 	const op = "perms.(Grant).MarshalJSON"
 	res := make(map[string]any, 4)
 	if g.id != "" {
@@ -203,7 +203,7 @@ func (g Grant) MarshalJSON(ctx context.Context) ([]byte, error) {
 	}
 	b, err := json.Marshal(res)
 	if err != nil {
-		return nil, errors.Wrap(ctx, err, op, errors.WithCode(errors.Encode))
+		return nil, fmt.Errorf("%s: error marshaling grant: %w", op, err)
 	}
 	return b, nil
 }

internal/perms/grants_test.go

@@ -132,8 +132,6 @@ func Test_ValidateType(t *testing.T) {
 func Test_MarshalingAndCloning(t *testing.T) {
 	t.Parallel()
 
-	ctx := context.Background()
-
 	type input struct {
 		name            string
 		input           Grant
@@ -278,7 +276,7 @@ func Test_MarshalingAndCloning(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			output, err := test.input.MarshalJSON(ctx)
+			output, err := test.input.MarshalJSON()
 			require.NoError(t, err)
 			assert.Equal(t, test.jsonOutput, string(output))
 			assert.Equal(t, test.canonicalString, test.input.CanonicalString())
@@ -1342,7 +1340,7 @@ func FuzzParse(f *testing.F) {
 		if g.CanonicalString() != g2.CanonicalString() {
 			t.Errorf("grant roundtrip failed, input %q, output %q", g.CanonicalString(), g2.CanonicalString())
 		}
-		jsonBytes, err := g.MarshalJSON(ctx)
+		jsonBytes, err := g.MarshalJSON()
 		if err != nil {
 			t.Error("Failed to marshal JSON:", err)
 		}

internal/plugin/loopback/testing_grpc_stream_test.go

@@ -415,7 +415,7 @@ func Test_GetObjectStream(t *testing.T) {
 		}, time.Second*10, time.Millisecond*50)
 
 		require.Eventually(func() bool {
-			return stream.streamClosed && stream.IsStreamClosed()
+			return stream.IsStreamClosed()
 		}, time.Second*10, time.Millisecond*50)
 	})
 }

internal/session/service_list_ext_test.go

@@ -155,7 +155,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, s *session.Session) (bool, error) {
 				return true, nil
 			}
-			_, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true)
+			_, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true)
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {

internal/target/service_list_ext_test.go

@@ -160,7 +160,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, t target.Target) (bool, error) {
 				return true, nil
 			}
-			_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
+			_, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {
@@ -238,7 +238,7 @@ func TestService_List(t *testing.T) {
 			filterFunc := func(_ context.Context, t target.Target) (bool, error) {
 				return true, nil
 			}
-			_, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
+			_, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
 			require.ErrorContains(t, err, "missing token")
 		})
 		t.Run("wrong token type", func(t *testing.T) {