Lock bucket while modifying its metadata (Fixes #2979) (#3019)

- When modifying notification configuration
- When modifying listener configuration
- When modifying policy configuration

With this change we also stop early checking if the bucket exists, since
that uses a Read-lock and causes a deadlock due to the outer Write-lock.

cmd/bucket-notification-handlers.go

@@ -164,6 +164,13 @@ func PutBucketNotificationConfig(bucket string, ncfg *notificationConfig, objAPI
 		return errInvalidArgument
 	}
 
+	// Acquire a write lock on bucket before modifying its
+	// configuration.
+	opsID := getOpsID()
+	nsMutex.Lock(bucket, "", opsID)
+	// Release lock after notifying peers
+	defer nsMutex.Unlock(bucket, "", opsID)
+
 	// persist config to disk
 	err := persistNotificationConfig(bucket, ncfg, objAPI)
 	if err != nil {
@@ -365,6 +372,13 @@ func AddBucketListenerConfig(bucket string, lcfg *listenerConfig, objAPI ObjectL
 	// add new lid to listeners and persist to object layer.
 	listenerCfgs = append(listenerCfgs, *lcfg)
 
+	// Acquire a write lock on bucket before modifying its
+	// configuration.
+	opsID := getOpsID()
+	nsMutex.Lock(bucket, "", opsID)
+	// Release lock after notifying peers
+	defer nsMutex.Unlock(bucket, "", opsID)
+
 	// update persistent config
 	err := persistListenerConfig(bucket, listenerCfgs, objAPI)
 	if err != nil {
@@ -398,6 +412,13 @@ func RemoveBucketListenerConfig(bucket string, lcfg *listenerConfig, objAPI Obje
 		return
 	}
 
+	// Acquire a write lock on bucket before modifying its
+	// configuration.
+	opsID := getOpsID()
+	nsMutex.Lock(bucket, "", opsID)
+	// Release lock after notifying peers
+	defer nsMutex.Unlock(bucket, "", opsID)
+
 	// update persistent config
 	err := persistListenerConfig(bucket, updatedLcfgs, objAPI)
 	if err != nil {