From 6cbfa2a3a569fcc856788b37616876538d904b3d Mon Sep 17 00:00:00 2001
From: Matt McShane <matt@mattmcshane.com>
Date: Wed, 20 Oct 2021 15:27:58 -0400
Subject: [PATCH] Avoid acquiring shared lock recursively (#2076)

Per the Go docs for sync.RWMutex, pending calls to RWMutex.Lock() block
calls to RWMutex.RLock(). With this understood we can see that the call
from getNamespace to getNamespaceByID acquires the read side of the
RWMutex twice - any interleaved call to RWMutex.Lock() between those two
RLock() calls will cause a deadlock - the Lock() call cannot continue
and the second RLock() call also will not continue.
---
 common/namespace/registry.go      |  6 +++++-
 common/namespace/registry_test.go | 26 ++++++++++++++++++++++++++
 2 files changed, 31 insertions(+), 1 deletion(-)

diff --git a/common/namespace/registry.go b/common/namespace/registry.go
index 40a035e037d..c12733cbc23 100644
--- a/common/namespace/registry.go
+++ b/common/namespace/registry.go
@@ -473,7 +473,7 @@ func (r *registry) getNamespace(name string) (*Namespace, error) {
 	r.cacheLock.RLock()
 	defer r.cacheLock.RUnlock()
 	if id, ok := r.cacheNameToID.Get(name).(string); ok {
-		return r.getNamespaceByID(id)
+		return r.getNamespaceByIDLocked(id)
 	}
 	return nil, serviceerror.NewNotFound(
 		fmt.Sprintf("Namespace name %q not found", name))
@@ -483,6 +483,10 @@ func (r *registry) getNamespace(name string) (*Namespace, error) {
 func (r *registry) getNamespaceByID(id string) (*Namespace, error) {
 	r.cacheLock.RLock()
 	defer r.cacheLock.RUnlock()
+	return r.getNamespaceByIDLocked(id)
+}
+
+func (r *registry) getNamespaceByIDLocked(id string) (*Namespace, error) {
 	if ns, ok := r.cacheByID.Get(id).(*Namespace); ok {
 		return ns, nil
 	}
diff --git a/common/namespace/registry_test.go b/common/namespace/registry_test.go
index b88852262c0..25fdb0981ed 100644
--- a/common/namespace/registry_test.go
+++ b/common/namespace/registry_test.go
@@ -528,3 +528,29 @@ func (s *registrySuite) TestGetTriggerListAndUpdateCache_ConcurrentAccess() {
 	close(startChan)
 	waitGroup.Wait()
 }
+
+func TestCacheByName(t *testing.T) {
+	nsrec := persistence.GetNamespaceResponse{
+		Namespace: &persistencespb.NamespaceDetail{
+			Info: &persistencespb.NamespaceInfo{
+				Id:   uuid.NewString(),
+				Name: "foo",
+			},
+			Config:            &persistencespb.NamespaceConfig{},
+			ReplicationConfig: &persistencespb.NamespaceReplicationConfig{},
+		},
+	}
+	regPersist := persistence.NewMockMetadataManager(gomock.NewController(t))
+	regPersist.EXPECT().GetMetadata().Return(
+		&persistence.GetMetadataResponse{NotificationVersion: nsrec.NotificationVersion + 1}, nil)
+	regPersist.EXPECT().ListNamespaces(gomock.Any()).Return(&persistence.ListNamespacesResponse{
+		Namespaces: []*persistence.GetNamespaceResponse{&nsrec},
+	}, nil)
+	reg := namespace.NewRegistry(
+		regPersist, false, metrics.NewNoopMetricsClient(), log.NewNoopLogger())
+	reg.Start()
+	defer reg.Stop()
+	ns, err := reg.GetNamespace("foo")
+	require.NoError(t, err)
+	require.Equal(t, "foo", ns.Name())
+}