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

@@ -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
 	}

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())
+}